[core] Remove all lineage implementation (#4607)

docs/layouts/shortcodes/generated/catalog_configuration.html

@@ -86,12 +86,6 @@
             <td>Boolean</td>
             <td>Whether to support format tables, format table corresponds to a regular csv, parquet or orc table, allowing read and write operations. However, during these processes, it does not connect to the metastore; hence, newly added partitions will not be reflected in the metastore and need to be manually added as separate partition operations.</td>
         </tr>
-        <tr>
-            <td><h5>lineage-meta</h5></td>
-            <td style="word-wrap: break-word;">(none)</td>
-            <td>String</td>
-            <td>The lineage meta to store table and data lineage information.<br /><br />Possible values:<br /><ul><li>"jdbc": Use standard jdbc to store table and data lineage information.</li></ul><ul><li>"custom": You can implement LineageMetaFactory and LineageMeta to store lineage information in customized storage.</li></ul></td>
-        </tr>
         <tr>
             <td><h5>lock-acquire-timeout</h5></td>
             <td style="word-wrap: break-word;">8 min</td>

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.
  *
  * <p>A factory is uniquely identified by {@link Class} and {@link #identifier()}.
  *

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<String> 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<Boolean> ALLOW_UPPER_CASE =
             ConfigOptions.key("allow-upper-case")
                     .booleanType()

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<String, String> 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<SchemaChange> 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();