OpenLineage : Add core lineage service scaffolding

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,71 @@
+/*
+ * 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 jakarta.annotation.Nullable;
+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,
+    @Nullable 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,
+      @Nullable String subType,
+      long createdAt,
+      long updatedAt) {
+    this(
+        OptionalLong.of(catalogId),
+        OptionalLong.of(datasetId),
+        namespace,
+        name,
+        subType,
+        OptionalLong.of(createdAt),
+        OptionalLong.of(updatedAt));
+  }
+
+  public LineageData(LineageDataset dataset) {
+    this(
+        OptionalLong.empty(),
+        dataset.polarisEntityId(),
+        dataset.namespace(),
+        dataset.name(),
+        null,
+        OptionalLong.empty(),
+        OptionalLong.empty());
+  }
+}

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,33 @@
+/*
+ * 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");
+  }
+
+  public LineageFieldMapping(LineageColumnEdge edge) {
+    this(edge.source().field(), edge.target().field());
+  }
+}

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(upstream);
+    downstream = List.copyOf(downstream);
+  }
+}

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(datasets);
+    edges = List.copyOf(edges);
+    columnEdges = List.copyOf(columnEdges);
+    Objects.requireNonNull(eventTime, "eventTime must be non-null");
+  }
+}