OpenLineage: add Persistence Layer SPI

polaris-core/src/main/java/org/apache/polaris/core/config/FeatureConfiguration.java

@@ -357,6 +357,13 @@ public static void enforceFeatureEnabledOrThrow(
           .defaultValue(true)
           .buildFeatureConfiguration();
 
+  public static final FeatureConfiguration<Boolean> ENABLE_LINEAGE =
+      PolarisConfiguration.<Boolean>builder()
+          .key("ENABLE_LINEAGE")
+          .description("If true, lineage services are enabled")
+          .defaultValue(false)
+          .buildFeatureConfiguration();
+
   public static final FeatureConfiguration<List<String>> SUPPORTED_CATALOG_CONNECTION_TYPES =
       PolarisConfiguration.<List<String>>builder()
           .key("SUPPORTED_CATALOG_CONNECTION_TYPES")

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageColumnEdge.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+
+/** A field-level lineage relationship between two dataset columns. */
+public record LineageColumnEdge(LineageFieldReference source, LineageFieldReference target) {
+  public LineageColumnEdge {
+    Objects.requireNonNull(source, "source must be non-null");
+    Objects.requireNonNull(target, "target must be non-null");
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageData.java

@@ -0,0 +1,59 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+import java.util.OptionalLong;
+
+/** Dataset metadata returned in a lineage query response. */
+public record LineageData(
+    OptionalLong catalogId,
+    OptionalLong datasetId,
+    String namespace,
+    String name,
+    String subType,
+    OptionalLong createdAt,
+    OptionalLong updatedAt) {
+  public LineageData {
+    Objects.requireNonNull(catalogId, "catalogId must be non-null");
+    Objects.requireNonNull(datasetId, "datasetId must be non-null");
+    Objects.requireNonNull(namespace, "namespace must be non-null");
+    Objects.requireNonNull(name, "name must be non-null");
+    Objects.requireNonNull(createdAt, "createdAt must be non-null");
+    Objects.requireNonNull(updatedAt, "updatedAt must be non-null");
+  }
+
+  public LineageData(
+      long catalogId,
+      long datasetId,
+      String namespace,
+      String name,
+      String subType,
+      long createdAt,
+      long updatedAt) {
+    this(
+        OptionalLong.of(catalogId),
+        OptionalLong.of(datasetId),
+        namespace,
+        name,
+        subType,
+        OptionalLong.of(createdAt),
+        OptionalLong.of(updatedAt));
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageDataset.java

@@ -0,0 +1,37 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+import java.util.OptionalLong;
+
+/** A dataset participating in lineage. */
+public record LineageDataset(
+    String catalog, String namespace, String name, OptionalLong polarisEntityId) {
+  public LineageDataset {
+    Objects.requireNonNull(catalog, "catalog must be non-null");
+    Objects.requireNonNull(namespace, "namespace must be non-null");
+    Objects.requireNonNull(name, "name must be non-null");
+    Objects.requireNonNull(polarisEntityId, "polarisEntityId must be non-null");
+  }
+
+  public LineageDataset(String catalog, String namespace, String name) {
+    this(catalog, namespace, name, OptionalLong.empty());
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageDirection.java

@@ -0,0 +1,26 @@
+/*
+ * 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.polaris.core.lineage;
+
+/** Supported directions for lineage queries. */
+public enum LineageDirection {
+  UPSTREAM,
+  DOWNSTREAM,
+  BOTH
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageEdge.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+
+/** A dataset-level lineage relationship. */
+public record LineageEdge(LineageDataset source, LineageDataset target) {
+  public LineageEdge {
+    Objects.requireNonNull(source, "source must be non-null");
+    Objects.requireNonNull(target, "target must be non-null");
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageFieldMapping.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+
+/** A source-to-target field mapping returned for column-granularity queries. */
+public record LineageFieldMapping(String sourceField, String targetField) {
+  public LineageFieldMapping {
+    Objects.requireNonNull(sourceField, "sourceField must be non-null");
+    Objects.requireNonNull(targetField, "targetField must be non-null");
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageFieldReference.java

@@ -0,0 +1,29 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.Objects;
+
+/** A reference to a specific field on a lineage dataset. */
+public record LineageFieldReference(LineageDataset dataset, String field) {
+  public LineageFieldReference {
+    Objects.requireNonNull(dataset, "dataset must be non-null");
+    Objects.requireNonNull(field, "field must be non-null");
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageGranularity.java

@@ -0,0 +1,25 @@
+/*
+ * 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.polaris.core.lineage;
+
+/** Supported query granularities for lineage lookups. */
+public enum LineageGranularity {
+  DATASET,
+  COLUMN
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageGraph.java

@@ -0,0 +1,32 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.List;
+import java.util.Objects;
+
+/** Normalized response model for lineage queries. */
+public record LineageGraph(
+    LineageNode node, List<LineageNode> upstream, List<LineageNode> downstream) {
+  public LineageGraph {
+    Objects.requireNonNull(node, "node must be non-null");
+    upstream = List.copyOf(Objects.requireNonNull(upstream, "upstream must be non-null"));
+    downstream = List.copyOf(Objects.requireNonNull(downstream, "downstream must be non-null"));
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageIngestRequest.java

@@ -0,0 +1,38 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.time.Instant;
+import java.util.List;
+import java.util.Objects;
+import java.util.Optional;
+
+/** Extracted lineage payload that can be persisted independent of the transport event shape. */
+public record LineageIngestRequest(
+    List<LineageDataset> datasets,
+    List<LineageEdge> edges,
+    List<LineageColumnEdge> columnEdges,
+    Optional<Instant> eventTime) {
+  public LineageIngestRequest {
+    datasets = List.copyOf(Objects.requireNonNull(datasets, "datasets must be non-null"));
+    edges = List.copyOf(Objects.requireNonNull(edges, "edges must be non-null"));
+    columnEdges = List.copyOf(Objects.requireNonNull(columnEdges, "columnEdges must be non-null"));
+    Objects.requireNonNull(eventTime, "eventTime must be non-null");
+  }
+}

polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageNode.java

@@ -0,0 +1,41 @@
+/*
+ * 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.polaris.core.lineage;
+
+import java.util.List;
+import java.util.Objects;
+
+/** A node returned in a lineage query response. */
+public record LineageNode(
+    String id,
+    LineageNodeType type,
+    LineageData data,
+    boolean opaque,
+    List<LineageFieldMapping> fieldMappings) {
+  public LineageNode {
+    Objects.requireNonNull(id, "id must be non-null");
+    Objects.requireNonNull(type, "type must be non-null");
+    fieldMappings =
+        List.copyOf(Objects.requireNonNull(fieldMappings, "fieldMappings must be non-null"));
+  }
+
+  public LineageNode(String id, LineageNodeType type, LineageData data, boolean opaque) {
+    this(id, type, data, opaque, List.of());
+  }
+}