lock-acquire-timeout |
8 min |
diff --git a/paimon-common/src/main/java/org/apache/paimon/factories/Factory.java b/paimon-common/src/main/java/org/apache/paimon/factories/Factory.java
index b0f1ec84c170..74796879ef4b 100644
--- a/paimon-common/src/main/java/org/apache/paimon/factories/Factory.java
+++ b/paimon-common/src/main/java/org/apache/paimon/factories/Factory.java
@@ -20,7 +20,7 @@
/**
* Base interface for all kind of factories that create object instances from a list of key-value
- * pairs in Paimon's catalog, lineage.
+ * pairs in Paimon's catalog.
*
* A factory is uniquely identified by {@link Class} and {@link #identifier()}.
*
diff --git a/paimon-common/src/main/java/org/apache/paimon/lineage/DataLineageEntity.java b/paimon-common/src/main/java/org/apache/paimon/lineage/DataLineageEntity.java
deleted file mode 100644
index e7401a9be3b7..000000000000
--- a/paimon-common/src/main/java/org/apache/paimon/lineage/DataLineageEntity.java
+++ /dev/null
@@ -1,33 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.lineage;
-
-import org.apache.paimon.data.Timestamp;
-
-/**
- * Data lineage entity with table lineage, barrier id and snapshot id for table source and sink
- * lineage.
- */
-public interface DataLineageEntity extends TableLineageEntity {
- long getBarrierId();
-
- long getSnapshotId();
-
- Timestamp getCreateTime();
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMeta.java b/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMeta.java
deleted file mode 100644
index 5d1c42daf6c8..000000000000
--- a/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMeta.java
+++ /dev/null
@@ -1,102 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.lineage;
-
-import org.apache.paimon.predicate.Predicate;
-
-import javax.annotation.Nullable;
-
-import java.util.Iterator;
-
-/** Metadata store will manage table lineage and data lineage information for the catalog. */
-public interface LineageMeta extends AutoCloseable {
- /**
- * Save the source table and job lineage.
- *
- * @param entity the table lineage entity
- */
- void saveSourceTableLineage(TableLineageEntity entity);
-
- /**
- * Delete the source table lineage for given job.
- *
- * @param job the job for table lineage
- */
- void deleteSourceTableLineage(String job);
-
- /**
- * Get source table and job lineages.
- *
- * @param predicate the predicate for the table lineages
- * @return the iterator for source table and job lineages
- */
- Iterator sourceTableLineages(@Nullable Predicate predicate);
-
- /**
- * Save the sink table and job lineage.
- *
- * @param entity the table lineage entity
- */
- void saveSinkTableLineage(TableLineageEntity entity);
-
- /**
- * Get sink table and job lineages.
- *
- * @param predicate the predicate for the table lineages
- * @return the iterator for sink table and job lineages
- */
- Iterator sinkTableLineages(@Nullable Predicate predicate);
-
- /**
- * Delete the sink table lineage for given job.
- *
- * @param job the job for table lineage
- */
- void deleteSinkTableLineage(String job);
-
- /**
- * Save the source table and job lineage.
- *
- * @param entity the data lineage entity
- */
- void saveSourceDataLineage(DataLineageEntity entity);
-
- /**
- * Get source data and job lineages.
- *
- * @param predicate the predicate for the table lineages
- * @return the iterator for source table and job lineages
- */
- Iterator sourceDataLineages(@Nullable Predicate predicate);
-
- /**
- * Save the sink table and job lineage.
- *
- * @param entity the data lineage entity
- */
- void saveSinkDataLineage(DataLineageEntity entity);
-
- /**
- * Get sink data and job lineages.
- *
- * @param predicate the predicate for the table lineages
- * @return the iterator for sink table and job lineages
- */
- Iterator sinkDataLineages(@Nullable Predicate predicate);
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMetaFactory.java b/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMetaFactory.java
deleted file mode 100644
index 11c6d3a1173c..000000000000
--- a/paimon-common/src/main/java/org/apache/paimon/lineage/LineageMetaFactory.java
+++ /dev/null
@@ -1,37 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.lineage;
-
-import org.apache.paimon.factories.Factory;
-import org.apache.paimon.options.Options;
-
-import java.io.Serializable;
-
-/** Factory to create {@link LineageMeta}. Each factory should have a unique identifier. */
-public interface LineageMetaFactory extends Factory, Serializable {
-
- LineageMeta create(LineageMetaContext context);
-
- /**
- * Context has all options in a catalog and is used in factory to create {@link LineageMeta}.
- */
- interface LineageMetaContext {
- Options options();
- }
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntity.java b/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntity.java
deleted file mode 100644
index c4312c4eb080..000000000000
--- a/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntity.java
+++ /dev/null
@@ -1,32 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.lineage;
-
-import org.apache.paimon.data.Timestamp;
-
-/** Table lineage entity with database, table and job for table source and sink lineage. */
-public interface TableLineageEntity {
- String getDatabase();
-
- String getTable();
-
- String getJob();
-
- Timestamp getCreateTime();
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntityImpl.java b/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntityImpl.java
deleted file mode 100644
index ef11ee87f15c..000000000000
--- a/paimon-common/src/main/java/org/apache/paimon/lineage/TableLineageEntityImpl.java
+++ /dev/null
@@ -1,56 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.lineage;
-
-import org.apache.paimon.data.Timestamp;
-
-/** Default implementation for {@link TableLineageEntity}. */
-public class TableLineageEntityImpl implements TableLineageEntity {
- private final String database;
- private final String table;
- private final String job;
- private final Timestamp timestamp;
-
- public TableLineageEntityImpl(String database, String table, String job, Timestamp timestamp) {
- this.database = database;
- this.table = table;
- this.job = job;
- this.timestamp = timestamp;
- }
-
- @Override
- public String getDatabase() {
- return database;
- }
-
- @Override
- public String getTable() {
- return table;
- }
-
- @Override
- public String getJob() {
- return job;
- }
-
- @Override
- public Timestamp getCreateTime() {
- return timestamp;
- }
-}
diff --git a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
index f69af2d59910..bb8cfae68284 100644
--- a/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
+++ b/paimon-common/src/main/java/org/apache/paimon/options/CatalogOptions.java
@@ -18,8 +18,6 @@
package org.apache.paimon.options;
-import org.apache.paimon.options.description.Description;
-import org.apache.paimon.options.description.TextElement;
import org.apache.paimon.table.CatalogTableType;
import java.time.Duration;
@@ -130,26 +128,6 @@ public class CatalogOptions {
.withDescription(
"Controls the max number for snapshots per table in the catalog are cached.");
- public static final ConfigOption LINEAGE_META =
- key("lineage-meta")
- .stringType()
- .noDefaultValue()
- .withDescription(
- Description.builder()
- .text(
- "The lineage meta to store table and data lineage information.")
- .linebreak()
- .linebreak()
- .text("Possible values:")
- .linebreak()
- .list(
- TextElement.text(
- "\"jdbc\": Use standard jdbc to store table and data lineage information."))
- .list(
- TextElement.text(
- "\"custom\": You can implement LineageMetaFactory and LineageMeta to store lineage information in customized storage."))
- .build());
-
public static final ConfigOption ALLOW_UPPER_CASE =
ConfigOptions.key("allow-upper-case")
.booleanType()
diff --git a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
index 16b76513d7ef..2b277a29b835 100644
--- a/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
+++ b/paimon-core/src/main/java/org/apache/paimon/catalog/AbstractCatalog.java
@@ -24,7 +24,6 @@
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.FileStatus;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.lineage.LineageMetaFactory;
import org.apache.paimon.manifest.PartitionEntry;
import org.apache.paimon.metastore.MetastoreClient;
import org.apache.paimon.operation.FileStoreCommit;
@@ -62,7 +61,6 @@
import static org.apache.paimon.CoreOptions.TYPE;
import static org.apache.paimon.CoreOptions.createCommitUser;
import static org.apache.paimon.options.CatalogOptions.ALLOW_UPPER_CASE;
-import static org.apache.paimon.options.CatalogOptions.LINEAGE_META;
import static org.apache.paimon.options.CatalogOptions.LOCK_ENABLED;
import static org.apache.paimon.options.CatalogOptions.LOCK_TYPE;
import static org.apache.paimon.utils.BranchManager.DEFAULT_MAIN_BRANCH;
@@ -76,19 +74,14 @@ public abstract class AbstractCatalog implements Catalog {
protected final Map tableDefaultOptions;
protected final Options catalogOptions;
- @Nullable protected final LineageMetaFactory lineageMetaFactory;
-
protected AbstractCatalog(FileIO fileIO) {
this.fileIO = fileIO;
- this.lineageMetaFactory = null;
this.tableDefaultOptions = new HashMap<>();
this.catalogOptions = new Options();
}
protected AbstractCatalog(FileIO fileIO, Options options) {
this.fileIO = fileIO;
- this.lineageMetaFactory =
- findAndCreateLineageMeta(options, AbstractCatalog.class.getClassLoader());
this.tableDefaultOptions = Catalog.tableDefaultOptions(options.toMap());
this.catalogOptions = options;
}
@@ -377,27 +370,13 @@ public void alterTable(
protected abstract void alterTableImpl(Identifier identifier, List changes)
throws TableNotExistException, ColumnAlreadyExistException, ColumnNotExistException;
- @Nullable
- private LineageMetaFactory findAndCreateLineageMeta(Options options, ClassLoader classLoader) {
- return options.getOptional(LINEAGE_META)
- .map(
- meta ->
- FactoryUtil.discoverFactory(
- classLoader, LineageMetaFactory.class, meta))
- .orElse(null);
- }
-
@Override
public Table getTable(Identifier identifier) throws TableNotExistException {
if (isSystemDatabase(identifier.getDatabaseName())) {
String tableName = identifier.getTableName();
Table table =
SystemTableLoader.loadGlobal(
- tableName,
- fileIO,
- this::allTablePaths,
- catalogOptions,
- lineageMetaFactory);
+ tableName, fileIO, this::allTablePaths, catalogOptions);
if (table == null) {
throw new TableNotExistException(identifier);
}
@@ -444,8 +423,7 @@ protected Table getDataOrFormatTable(Identifier identifier) throws TableNotExist
lockFactory().orElse(null),
lockContext().orElse(null),
identifier),
- metastoreClientFactory(identifier, tableMeta.schema).orElse(null),
- lineageMetaFactory));
+ metastoreClientFactory(identifier, tableMeta.schema).orElse(null)));
CoreOptions options = table.coreOptions();
if (options.type() == TableType.OBJECT_TABLE) {
String objectLocation = options.objectLocation();
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
index 9ff5f9b4f6a8..a722d9e21ada 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/CatalogEnvironment.java
@@ -19,7 +19,6 @@
package org.apache.paimon.table;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.lineage.LineageMetaFactory;
import org.apache.paimon.metastore.MetastoreClient;
import org.apache.paimon.operation.Lock;
@@ -27,10 +26,7 @@
import java.io.Serializable;
-/**
- * Catalog environment in table which contains log factory, metastore client factory and lineage
- * meta.
- */
+/** Catalog environment in table which contains log factory, metastore client factory. */
public class CatalogEnvironment implements Serializable {
private static final long serialVersionUID = 1L;
@@ -39,23 +35,20 @@ public class CatalogEnvironment implements Serializable {
@Nullable private final String uuid;
private final Lock.Factory lockFactory;
@Nullable private final MetastoreClient.Factory metastoreClientFactory;
- @Nullable private final LineageMetaFactory lineageMetaFactory;
public CatalogEnvironment(
@Nullable Identifier identifier,
@Nullable String uuid,
Lock.Factory lockFactory,
- @Nullable MetastoreClient.Factory metastoreClientFactory,
- @Nullable LineageMetaFactory lineageMetaFactory) {
+ @Nullable MetastoreClient.Factory metastoreClientFactory) {
this.identifier = identifier;
this.uuid = uuid;
this.lockFactory = lockFactory;
this.metastoreClientFactory = metastoreClientFactory;
- this.lineageMetaFactory = lineageMetaFactory;
}
public static CatalogEnvironment empty() {
- return new CatalogEnvironment(null, null, Lock.emptyFactory(), null, null);
+ return new CatalogEnvironment(null, null, Lock.emptyFactory(), null);
}
@Nullable
@@ -76,9 +69,4 @@ public Lock.Factory lockFactory() {
public MetastoreClient.Factory metastoreClientFactory() {
return metastoreClientFactory;
}
-
- @Nullable
- public LineageMetaFactory lineageMetaFactory() {
- return lineageMetaFactory;
- }
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SinkTableLineageTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SinkTableLineageTable.java
deleted file mode 100644
index 71efce070471..000000000000
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SinkTableLineageTable.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.table.system;
-
-import org.apache.paimon.lineage.LineageMeta;
-import org.apache.paimon.lineage.LineageMetaFactory;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.InnerTableRead;
-
-import java.util.Map;
-
-/**
- * This is a system table to display all the sink table lineages.
- *
- *
- * For example:
- * If we select * from sys.sink_table_lineage, we will get
- * database_name table_name job_name create_time
- * default test0 job1 2023-10-22 20:35:12
- * database1 test1 job1 2023-10-28 21:35:52
- * ... ... ... ...
- * We can write sql to fetch the information we need.
- *
- */
-public class SinkTableLineageTable extends TableLineageTable {
-
- public static final String SINK_TABLE_LINEAGE = "sink_table_lineage";
-
- public SinkTableLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
- super(lineageMetaFactory, options);
- }
-
- @Override
- public InnerTableRead newRead() {
- return new TableLineageRead(lineageMetaFactory, options, LineageMeta::sinkTableLineages);
- }
-
- @Override
- public String name() {
- return SINK_TABLE_LINEAGE;
- }
-
- @Override
- public Table copy(Map dynamicOptions) {
- return new SinkTableLineageTable(lineageMetaFactory, options);
- }
-}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SourceTableLineageTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SourceTableLineageTable.java
deleted file mode 100644
index 5d9904fa6675..000000000000
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SourceTableLineageTable.java
+++ /dev/null
@@ -1,64 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.table.system;
-
-import org.apache.paimon.lineage.LineageMeta;
-import org.apache.paimon.lineage.LineageMetaFactory;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.table.Table;
-import org.apache.paimon.table.source.InnerTableRead;
-
-import java.util.Map;
-
-/**
- * This is a system table to display all the source table lineages.
- *
- *
- * For example:
- * If we select * from sys.source_table_lineage, we will get
- * database_name table_name job_name create_time
- * default test0 job1 2023-10-22 20:35:12
- * database1 test1 job1 2023-10-28 21:35:52
- * ... ... ... ...
- * We can write sql to fetch the information we need.
- *
- */
-public class SourceTableLineageTable extends TableLineageTable {
-
- public static final String SOURCE_TABLE_LINEAGE = "source_table_lineage";
-
- public SourceTableLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
- super(lineageMetaFactory, options);
- }
-
- @Override
- public InnerTableRead newRead() {
- return new TableLineageRead(lineageMetaFactory, options, LineageMeta::sourceTableLineages);
- }
-
- @Override
- public String name() {
- return SOURCE_TABLE_LINEAGE;
- }
-
- @Override
- public Table copy(Map dynamicOptions) {
- return new SourceTableLineageTable(lineageMetaFactory, options);
- }
-}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
index 3d5b211316ec..763e4d121673 100644
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
+++ b/paimon-core/src/main/java/org/apache/paimon/table/system/SystemTableLoader.java
@@ -20,7 +20,6 @@
import org.apache.paimon.fs.FileIO;
import org.apache.paimon.fs.Path;
-import org.apache.paimon.lineage.LineageMetaFactory;
import org.apache.paimon.options.Options;
import org.apache.paimon.table.FileStoreTable;
import org.apache.paimon.table.Table;
@@ -37,7 +36,6 @@
import java.util.function.Function;
import java.util.function.Supplier;
-import static org.apache.paimon.options.CatalogOptions.LINEAGE_META;
import static org.apache.paimon.table.system.AggregationFieldsTable.AGGREGATION_FIELDS;
import static org.apache.paimon.table.system.AllTableOptionsTable.ALL_TABLE_OPTIONS;
import static org.apache.paimon.table.system.AuditLogTable.AUDIT_LOG;
@@ -52,12 +50,9 @@
import static org.apache.paimon.table.system.PartitionsTable.PARTITIONS;
import static org.apache.paimon.table.system.ReadOptimizedTable.READ_OPTIMIZED;
import static org.apache.paimon.table.system.SchemasTable.SCHEMAS;
-import static org.apache.paimon.table.system.SinkTableLineageTable.SINK_TABLE_LINEAGE;
import static org.apache.paimon.table.system.SnapshotsTable.SNAPSHOTS;
-import static org.apache.paimon.table.system.SourceTableLineageTable.SOURCE_TABLE_LINEAGE;
import static org.apache.paimon.table.system.StatisticTable.STATISTICS;
import static org.apache.paimon.table.system.TagsTable.TAGS;
-import static org.apache.paimon.utils.Preconditions.checkNotNull;
/** Loader to load system {@link Table}s. */
public class SystemTableLoader {
@@ -95,38 +90,18 @@ public static Table loadGlobal(
String tableName,
FileIO fileIO,
Supplier>> allTablePaths,
- Options catalogOptions,
- @Nullable LineageMetaFactory lineageMetaFactory) {
+ Options catalogOptions) {
switch (tableName.toLowerCase()) {
case ALL_TABLE_OPTIONS:
return new AllTableOptionsTable(fileIO, allTablePaths.get());
case CATALOG_OPTIONS:
return new CatalogOptionsTable(catalogOptions);
- case SOURCE_TABLE_LINEAGE:
- {
- checkNotNull(
- lineageMetaFactory,
- String.format(
- "Lineage meta should be configured for catalog with %s",
- LINEAGE_META.key()));
- return new SourceTableLineageTable(lineageMetaFactory, catalogOptions);
- }
- case SINK_TABLE_LINEAGE:
- {
- checkNotNull(
- lineageMetaFactory,
- String.format(
- "Lineage meta should be configured for catalog with %s",
- LINEAGE_META.key()));
- return new SinkTableLineageTable(lineageMetaFactory, catalogOptions);
- }
default:
return null;
}
}
public static List loadGlobalTableNames() {
- return Arrays.asList(
- ALL_TABLE_OPTIONS, CATALOG_OPTIONS, SOURCE_TABLE_LINEAGE, SINK_TABLE_LINEAGE);
+ return Arrays.asList(ALL_TABLE_OPTIONS, CATALOG_OPTIONS);
}
}
diff --git a/paimon-core/src/main/java/org/apache/paimon/table/system/TableLineageTable.java b/paimon-core/src/main/java/org/apache/paimon/table/system/TableLineageTable.java
deleted file mode 100644
index aeaf3ca3b133..000000000000
--- a/paimon-core/src/main/java/org/apache/paimon/table/system/TableLineageTable.java
+++ /dev/null
@@ -1,168 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.table.system;
-
-import org.apache.paimon.data.BinaryString;
-import org.apache.paimon.data.GenericRow;
-import org.apache.paimon.data.InternalRow;
-import org.apache.paimon.disk.IOManager;
-import org.apache.paimon.lineage.LineageMeta;
-import org.apache.paimon.lineage.LineageMetaFactory;
-import org.apache.paimon.lineage.TableLineageEntity;
-import org.apache.paimon.options.Options;
-import org.apache.paimon.predicate.Predicate;
-import org.apache.paimon.reader.RecordReader;
-import org.apache.paimon.table.ReadonlyTable;
-import org.apache.paimon.table.source.InnerTableRead;
-import org.apache.paimon.table.source.InnerTableScan;
-import org.apache.paimon.table.source.ReadOnceTableScan;
-import org.apache.paimon.table.source.Split;
-import org.apache.paimon.table.source.TableRead;
-import org.apache.paimon.types.DataField;
-import org.apache.paimon.types.RowType;
-import org.apache.paimon.types.TimestampType;
-import org.apache.paimon.types.VarCharType;
-import org.apache.paimon.utils.IteratorRecordReader;
-import org.apache.paimon.utils.ProjectedRow;
-
-import org.apache.paimon.shade.guava30.com.google.common.collect.Iterators;
-
-import javax.annotation.Nullable;
-
-import java.io.IOException;
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.List;
-import java.util.function.BiFunction;
-
-import static org.apache.paimon.utils.Preconditions.checkNotNull;
-
-/** Base lineage table for source and sink table lineage. */
-public abstract class TableLineageTable implements ReadonlyTable {
- protected final LineageMetaFactory lineageMetaFactory;
- protected final Options options;
-
- public static final RowType TABLE_TYPE =
- new RowType(
- Arrays.asList(
- new DataField(
- 0, "database_name", new VarCharType(VarCharType.MAX_LENGTH)),
- new DataField(1, "table_name", new VarCharType(VarCharType.MAX_LENGTH)),
- new DataField(2, "job_name", new VarCharType(VarCharType.MAX_LENGTH)),
- new DataField(3, "create_time", new TimestampType())));
-
- protected TableLineageTable(LineageMetaFactory lineageMetaFactory, Options options) {
- this.lineageMetaFactory = lineageMetaFactory;
- this.options = options;
- }
-
- @Override
- public InnerTableScan newScan() {
- return new ReadOnceTableScan() {
- @Override
- public InnerTableScan withFilter(Predicate predicate) {
- return this;
- }
-
- @Override
- protected Plan innerPlan() {
- /// TODO get the real row count for plan.
- return () -> Collections.singletonList((Split) () -> 1L);
- }
- };
- }
-
- @Override
- public RowType rowType() {
- return TABLE_TYPE;
- }
-
- @Override
- public List primaryKeys() {
- return Arrays.asList("database_name", "table_name", "job_name");
- }
-
- /** Table lineage read with lineage meta query. */
- protected static class TableLineageRead implements InnerTableRead {
- private final LineageMetaFactory lineageMetaFactory;
- private final Options options;
- private final BiFunction>
- tableLineageQuery;
- @Nullable private Predicate predicate;
- private RowType readType;
-
- protected TableLineageRead(
- LineageMetaFactory lineageMetaFactory,
- Options options,
- BiFunction>
- tableLineageQuery) {
- this.lineageMetaFactory = lineageMetaFactory;
- this.options = options;
- this.tableLineageQuery = tableLineageQuery;
- this.predicate = null;
- }
-
- @Override
- public InnerTableRead withFilter(Predicate predicate) {
- this.predicate = predicate;
- return this;
- }
-
- @Override
- public InnerTableRead withReadType(RowType readType) {
- this.readType = readType;
- return this;
- }
-
- @Override
- public TableRead withIOManager(IOManager ioManager) {
- return this;
- }
-
- @Override
- public RecordReader createReader(Split split) throws IOException {
- try (LineageMeta lineageMeta = lineageMetaFactory.create(() -> options)) {
- Iterator sourceTableLineages =
- tableLineageQuery.apply(lineageMeta, predicate);
- return new IteratorRecordReader<>(
- Iterators.transform(
- sourceTableLineages,
- entity -> {
- checkNotNull(entity);
- GenericRow row =
- GenericRow.of(
- BinaryString.fromString(entity.getDatabase()),
- BinaryString.fromString(entity.getTable()),
- BinaryString.fromString(entity.getJob()),
- entity.getCreateTime());
- if (readType != null) {
- return ProjectedRow.from(
- readType, TableLineageTable.TABLE_TYPE)
- .replaceRow(row);
- } else {
- return row;
- }
- }));
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
- }
-}
diff --git a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
index 9f90a2cd0130..6b10dbb84bf4 100644
--- a/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
+++ b/paimon-flink/paimon-flink-common/src/main/java/org/apache/paimon/flink/AbstractFlinkTableFactory.java
@@ -25,15 +25,10 @@
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.catalog.CatalogContext;
import org.apache.paimon.catalog.Identifier;
-import org.apache.paimon.data.Timestamp;
import org.apache.paimon.flink.log.LogStoreTableFactory;
import org.apache.paimon.flink.sink.FlinkTableSink;
import org.apache.paimon.flink.source.DataTableSource;
import org.apache.paimon.flink.source.SystemTableSource;
-import org.apache.paimon.lineage.LineageMeta;
-import org.apache.paimon.lineage.LineageMetaFactory;
-import org.apache.paimon.lineage.TableLineageEntity;
-import org.apache.paimon.lineage.TableLineageEntityImpl;
import org.apache.paimon.options.Options;
import org.apache.paimon.options.OptionsUtils;
import org.apache.paimon.schema.Schema;
@@ -47,7 +42,6 @@
import org.apache.flink.configuration.ConfigOption;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.ExecutionOptions;
-import org.apache.flink.configuration.PipelineOptions;
import org.apache.flink.configuration.ReadableConfig;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.ValidationException;
@@ -71,7 +65,6 @@
import java.util.Objects;
import java.util.Optional;
import java.util.Set;
-import java.util.function.BiConsumer;
import java.util.regex.Pattern;
import static org.apache.paimon.CoreOptions.LOG_CHANGELOG_MODE;
@@ -109,23 +102,9 @@ public DynamicTableSource createDynamicTableSource(Context context) {
isStreamingMode,
context.getObjectIdentifier());
} else {
- Table table = buildPaimonTable(context);
- if (table instanceof FileStoreTable) {
- storeTableLineage(
- ((FileStoreTable) table).catalogEnvironment().lineageMetaFactory(),
- context,
- (entity, lineageFactory) -> {
- try (LineageMeta lineage =
- lineageFactory.create(() -> Options.fromMap(table.options()))) {
- lineage.saveSourceTableLineage(entity);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- }
return new DataTableSource(
context.getObjectIdentifier(),
- table,
+ buildPaimonTable(context),
isStreamingMode,
context,
createOptionalLogStoreFactory(context).orElse(null));
@@ -134,46 +113,13 @@ public DynamicTableSource createDynamicTableSource(Context context) {
@Override
public DynamicTableSink createDynamicTableSink(Context context) {
- Table table = buildPaimonTable(context);
- if (table instanceof FileStoreTable) {
- storeTableLineage(
- ((FileStoreTable) table).catalogEnvironment().lineageMetaFactory(),
- context,
- (entity, lineageFactory) -> {
- try (LineageMeta lineage =
- lineageFactory.create(() -> Options.fromMap(table.options()))) {
- lineage.saveSinkTableLineage(entity);
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- });
- }
return new FlinkTableSink(
context.getObjectIdentifier(),
- table,
+ buildPaimonTable(context),
context,
createOptionalLogStoreFactory(context).orElse(null));
}
- private void storeTableLineage(
- @Nullable LineageMetaFactory lineageMetaFactory,
- Context context,
- BiConsumer tableLineage) {
- if (lineageMetaFactory != null) {
- String pipelineName = context.getConfiguration().get(PipelineOptions.NAME);
- if (pipelineName == null) {
- throw new ValidationException("Cannot get pipeline name for lineage meta.");
- }
- tableLineage.accept(
- new TableLineageEntityImpl(
- context.getObjectIdentifier().getDatabaseName(),
- context.getObjectIdentifier().getObjectName(),
- pipelineName,
- Timestamp.fromEpochMillis(System.currentTimeMillis())),
- lineageMetaFactory);
- }
- }
-
@Override
public Set> requiredOptions() {
return Collections.emptySet();
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
index 96334de3f87b..10b03b7139ae 100644
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
+++ b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/CatalogTableITCase.java
@@ -21,8 +21,6 @@
import org.apache.paimon.catalog.Catalog;
import org.apache.paimon.table.system.AllTableOptionsTable;
import org.apache.paimon.table.system.CatalogOptionsTable;
-import org.apache.paimon.table.system.SinkTableLineageTable;
-import org.apache.paimon.table.system.SourceTableLineageTable;
import org.apache.paimon.utils.BlockingIterator;
import org.apache.commons.lang3.StringUtils;
@@ -200,9 +198,7 @@ public void testSystemDatabase() {
assertThat(sql("SHOW TABLES"))
.containsExactlyInAnyOrder(
Row.of(AllTableOptionsTable.ALL_TABLE_OPTIONS),
- Row.of(CatalogOptionsTable.CATALOG_OPTIONS),
- Row.of(SourceTableLineageTable.SOURCE_TABLE_LINEAGE),
- Row.of(SinkTableLineageTable.SINK_TABLE_LINEAGE));
+ Row.of(CatalogOptionsTable.CATALOG_OPTIONS));
}
@Test
diff --git a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkLineageITCase.java b/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkLineageITCase.java
deleted file mode 100644
index 5b61d5272f80..000000000000
--- a/paimon-flink/paimon-flink-common/src/test/java/org/apache/paimon/flink/FlinkLineageITCase.java
+++ /dev/null
@@ -1,206 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- *
- * http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-
-package org.apache.paimon.flink;
-
-import org.apache.paimon.lineage.DataLineageEntity;
-import org.apache.paimon.lineage.LineageMeta;
-import org.apache.paimon.lineage.LineageMetaFactory;
-import org.apache.paimon.lineage.TableLineageEntity;
-import org.apache.paimon.predicate.Predicate;
-
-import org.apache.flink.configuration.PipelineOptions;
-import org.apache.flink.table.api.ValidationException;
-import org.apache.flink.types.Row;
-import org.apache.flink.util.CloseableIterator;
-import org.junit.jupiter.api.Test;
-
-import javax.annotation.Nullable;
-
-import java.util.ArrayList;
-import java.util.Collections;
-import java.util.HashMap;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-
-import static org.apache.paimon.options.CatalogOptions.LINEAGE_META;
-import static org.assertj.core.api.Assertions.assertThat;
-import static org.assertj.core.api.Assertions.assertThatThrownBy;
-
-/** ITCase for flink table and data lineage. */
-public class FlinkLineageITCase extends CatalogITCaseBase {
- private static final String THROWING_META = "throwing-meta";
- private static final Map> jobSourceTableLineages =
- new HashMap<>();
- private static final Map> jobSinkTableLineages =
- new HashMap<>();
-
- @Override
- protected List ddl() {
- return Collections.singletonList("CREATE TABLE IF NOT EXISTS T (a INT, b INT, c INT)");
- }
-
- @Override
- protected Map catalogOptions() {
- return Collections.singletonMap(LINEAGE_META.key(), THROWING_META);
- }
-
- @Test
- public void testTableLineage() throws Exception {
- // Validate for source and sink lineage when pipeline name is null
- assertThatThrownBy(
- () -> tEnv.executeSql("INSERT INTO T VALUES (1, 2, 3),(4, 5, 6);").await())
- .hasCauseExactlyInstanceOf(ValidationException.class)
- .hasRootCauseMessage("Cannot get pipeline name for lineage meta.");
- assertThatThrownBy(() -> tEnv.executeSql("SELECT * FROM T").collect().close())
- .hasCauseExactlyInstanceOf(ValidationException.class)
- .hasRootCauseMessage("Cannot get pipeline name for lineage meta.");
-
- // Call storeSinkTableLineage and storeSourceTableLineage methods
- tEnv.getConfig().getConfiguration().set(PipelineOptions.NAME, "insert_t_job");
- tEnv.executeSql("INSERT INTO T VALUES (1, 2, 3),(4, 5, 6);").await();
- assertThat(jobSinkTableLineages).isNotEmpty();
- TableLineageEntity sinkTableLineage =
- jobSinkTableLineages.get("insert_t_job").get("default.T.insert_t_job");
- assertThat(sinkTableLineage.getTable()).isEqualTo("T");
-
- List sinkTableRows = new ArrayList<>();
- try (CloseableIterator iterator =
- tEnv.executeSql("SELECT * FROM sys.sink_table_lineage").collect()) {
- while (iterator.hasNext()) {
- sinkTableRows.add(iterator.next());
- }
- }
- assertThat(sinkTableRows.size()).isEqualTo(1);
- Row sinkTableRow = sinkTableRows.get(0);
- assertThat(sinkTableRow.getField("database_name")).isEqualTo("default");
- assertThat(sinkTableRow.getField("table_name")).isEqualTo("T");
- assertThat(sinkTableRow.getField("job_name")).isEqualTo("insert_t_job");
-
- tEnv.getConfig().getConfiguration().set(PipelineOptions.NAME, "select_t_job");
- tEnv.executeSql("SELECT * FROM T").collect().close();
- assertThat(jobSourceTableLineages).isNotEmpty();
- TableLineageEntity sourceTableLineage =
- jobSourceTableLineages.get("select_t_job").get("default.T.select_t_job");
- assertThat(sourceTableLineage.getTable()).isEqualTo("T");
-
- List sourceTableRows = new ArrayList<>();
- try (CloseableIterator iterator =
- tEnv.executeSql("SELECT * FROM sys.source_table_lineage").collect()) {
- while (iterator.hasNext()) {
- sourceTableRows.add(iterator.next());
- }
- }
- assertThat(sourceTableRows.size()).isEqualTo(1);
- Row sourceTableRow = sourceTableRows.get(0);
- assertThat(sourceTableRow.getField("database_name")).isEqualTo("default");
- assertThat(sourceTableRow.getField("table_name")).isEqualTo("T");
- assertThat(sourceTableRow.getField("job_name")).isEqualTo("select_t_job");
- }
-
- private static String getTableLineageKey(TableLineageEntity entity) {
- return String.format("%s.%s.%s", entity.getDatabase(), entity.getTable(), entity.getJob());
- }
-
- /** Factory to create throwing lineage meta. */
- public static class TestingMemoryLineageMetaFactory implements LineageMetaFactory {
- private static final long serialVersionUID = 1L;
-
- @Override
- public String identifier() {
- return THROWING_META;
- }
-
- @Override
- public LineageMeta create(LineageMetaContext context) {
- return new TestingMemoryLineageMeta();
- }
- }
-
- /** Throwing specific exception in each method. */
- private static class TestingMemoryLineageMeta implements LineageMeta {
-
- @Override
- public void saveSourceTableLineage(TableLineageEntity entity) {
- jobSourceTableLineages
- .computeIfAbsent(entity.getJob(), key -> new HashMap<>())
- .put(getTableLineageKey(entity), entity);
- }
-
- @Override
- public void deleteSourceTableLineage(String job) {
- jobSourceTableLineages.remove(job);
- }
-
- @Override
- public Iterator sourceTableLineages(@Nullable Predicate predicate) {
- return jobSourceTableLineages.values().stream()
- .flatMap(v -> v.values().stream())
- .iterator();
- }
-
- @Override
- public void saveSinkTableLineage(TableLineageEntity entity) {
- assertThat(entity.getJob()).isEqualTo("insert_t_job");
- assertThat(entity.getTable()).isEqualTo("T");
- assertThat(entity.getDatabase()).isEqualTo("default");
- jobSinkTableLineages
- .computeIfAbsent(entity.getJob(), key -> new HashMap<>())
- .put(getTableLineageKey(entity), entity);
- }
-
- @Override
- public Iterator sinkTableLineages(@Nullable Predicate predicate) {
- return jobSinkTableLineages.values().stream()
- .flatMap(v -> v.values().stream())
- .iterator();
- }
-
- @Override
- public void deleteSinkTableLineage(String job) {
- jobSinkTableLineages.remove(job);
- }
-
- @Override
- public void saveSourceDataLineage(DataLineageEntity entity) {
- assertThat(entity.getJob()).isEqualTo("select_t_job");
- assertThat(entity.getTable()).isEqualTo("T");
- assertThat(entity.getDatabase()).isEqualTo("default");
- throw new UnsupportedOperationException("Method saveSinkTableLineage is not supported");
- }
-
- @Override
- public Iterator sourceDataLineages(@Nullable Predicate predicate) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void saveSinkDataLineage(DataLineageEntity entity) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public Iterator sinkDataLineages(@Nullable Predicate predicate) {
- throw new UnsupportedOperationException();
- }
-
- @Override
- public void close() throws Exception {}
- }
-}
diff --git a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
index fcb6fe982943..3c05b5fba3ec 100644
--- a/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
+++ b/paimon-flink/paimon-flink-common/src/test/resources/META-INF/services/org.apache.paimon.factories.Factory
@@ -15,8 +15,5 @@
org.apache.paimon.flink.FlinkCatalogTest$TestingLogSoreRegisterFactory
-# Lineage meta factory
-org.apache.paimon.flink.FlinkLineageITCase$TestingMemoryLineageMetaFactory
-
# Catalog lock factory
org.apache.paimon.flink.FileSystemCatalogITCase$FileSystemCatalogDummyLockFactory
\ No newline at end of file
diff --git a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
index 9a90995f282d..5157e606006c 100644
--- a/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
+++ b/paimon-hive/paimon-hive-catalog/src/main/java/org/apache/paimon/hive/HiveCatalog.java
@@ -610,8 +610,7 @@ public org.apache.paimon.table.Table getDataOrFormatTable(Identifier identifier)
lockFactory().orElse(null),
lockContext().orElse(null),
identifier),
- metastoreClientFactory(identifier, tableMeta.schema()).orElse(null),
- lineageMetaFactory));
+ metastoreClientFactory(identifier, tableMeta.schema()).orElse(null)));
} catch (TableNotExistException ignore) {
}