Copilot commented on code in PR #4705: URL: https://github.com/apache/polaris/pull/4705#discussion_r3396171135
########## runtime/service/src/test/java/org/apache/polaris/service/lineage/DefaultLineageServiceTest.java: ########## @@ -0,0 +1,169 @@ +/* + * 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.service.lineage; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.OptionalLong; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.lineage.LineageColumnEdge; +import org.apache.polaris.core.lineage.LineageDataset; +import org.apache.polaris.core.lineage.LineageEdge; +import org.apache.polaris.core.lineage.LineageFieldReference; +import org.apache.polaris.core.lineage.LineageGraph; +import org.apache.polaris.core.lineage.LineageIngestRequest; +import org.apache.polaris.core.lineage.LineageNode; +import org.apache.polaris.core.lineage.LineageNodeType; +import org.apache.polaris.core.lineage.LineagePersistence; +import org.apache.polaris.core.lineage.LineageQueryRequest; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class DefaultLineageServiceTest { + private static final Instant EVENT_TIME = Instant.parse("2026-01-01T00:00:00Z"); + + @Mock private CallContext callContext; + @Mock private RealmContext realmContext; + @Mock private RealmConfig realmConfig; + @Mock private LineageConfiguration configuration; + @Mock private LineageConfiguration.PersistenceConfiguration persistenceConfiguration; + @Mock private LineagePersistence persistence; + + private DefaultLineageService service; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); Review Comment: MockitoAnnotations.openMocks(this) returns an AutoCloseable that should be closed to avoid leaking mock resources across tests. Prefer using JUnit 5’s MockitoExtension (@ExtendWith(MockitoExtension.class)) or store the returned Closeable and close it in an @AfterEach. ########## runtime/service/src/test/java/org/apache/polaris/service/lineage/DefaultLineageServiceTest.java: ########## @@ -0,0 +1,169 @@ +/* + * 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.service.lineage; + +import static org.assertj.core.api.Assertions.assertThatThrownBy; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.verifyNoInteractions; +import static org.mockito.Mockito.when; + +import java.time.Instant; +import java.util.List; +import java.util.Optional; +import java.util.OptionalLong; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.core.config.RealmConfig; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.lineage.LineageColumnEdge; +import org.apache.polaris.core.lineage.LineageDataset; +import org.apache.polaris.core.lineage.LineageEdge; +import org.apache.polaris.core.lineage.LineageFieldReference; +import org.apache.polaris.core.lineage.LineageGraph; +import org.apache.polaris.core.lineage.LineageIngestRequest; +import org.apache.polaris.core.lineage.LineageNode; +import org.apache.polaris.core.lineage.LineageNodeType; +import org.apache.polaris.core.lineage.LineagePersistence; +import org.apache.polaris.core.lineage.LineageQueryRequest; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.mockito.Mock; +import org.mockito.MockitoAnnotations; + +public class DefaultLineageServiceTest { + private static final Instant EVENT_TIME = Instant.parse("2026-01-01T00:00:00Z"); + + @Mock private CallContext callContext; + @Mock private RealmContext realmContext; + @Mock private RealmConfig realmConfig; + @Mock private LineageConfiguration configuration; + @Mock private LineageConfiguration.PersistenceConfiguration persistenceConfiguration; + @Mock private LineagePersistence persistence; + + private DefaultLineageService service; + + @BeforeEach + void setUp() { + MockitoAnnotations.openMocks(this); + when(callContext.getRealmContext()).thenReturn(realmContext); + when(callContext.getRealmConfig()).thenReturn(realmConfig); + when(configuration.persistence()).thenReturn(persistenceConfiguration); + service = new DefaultLineageService(callContext, configuration, persistence); + } + + @Test + void throwsWhenStaticConfigDisabled() { + when(configuration.enabled()).thenReturn(false); + + assertThatThrownBy(() -> service.ingest(emptyIngestRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("polaris.lineage.enabled"); + + assertThatThrownBy(() -> service.query(queryRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("polaris.lineage.enabled"); + + verifyNoInteractions(persistence); + } + + @Test + void throwsWhenRealmFeatureDisabled() { + when(configuration.enabled()).thenReturn(true); + when(realmConfig.getConfig(FeatureConfiguration.ENABLE_LINEAGE)).thenReturn(false); + + assertThatThrownBy(() -> service.ingest(emptyIngestRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining(FeatureConfiguration.ENABLE_LINEAGE.key()); + + assertThatThrownBy(() -> service.query(queryRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining(FeatureConfiguration.ENABLE_LINEAGE.key()); + + verifyNoInteractions(persistence); + } + + @Test + void throwsWhenPersistenceDisabled() { + when(configuration.enabled()).thenReturn(true); + when(realmConfig.getConfig(FeatureConfiguration.ENABLE_LINEAGE)).thenReturn(true); + when(persistenceConfiguration.enabled()).thenReturn(false); + + assertThatThrownBy(() -> service.ingest(emptyIngestRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("polaris.lineage.persistence.enabled"); + + assertThatThrownBy(() -> service.query(queryRequest())) + .isInstanceOf(UnsupportedOperationException.class) + .hasMessageContaining("polaris.lineage.persistence.enabled"); + + verifyNoInteractions(persistence); + } + + @Test + void delegatesWhenLineageEnabled() { + LineageIngestRequest ingestRequest = ingestRequest(); + LineageGraph graph = + new LineageGraph( + new LineageNode( + "dataset:test:orders", LineageNodeType.DATASET, dataset("test", "orders"), false), + List.of(), + List.of()); + + when(configuration.enabled()).thenReturn(true); + when(realmConfig.getConfig(FeatureConfiguration.ENABLE_LINEAGE)).thenReturn(true); + when(persistenceConfiguration.enabled()).thenReturn(true); + when(persistence.loadLineage(realmContext, queryRequest())).thenReturn(graph); + + service.ingest(ingestRequest); + service.query(queryRequest()); Review Comment: This test stubs persistence.loadLineage(...) but doesn’t assert that DefaultLineageService.query(...) returns the stubbed graph. Add an assertion on the returned value (in addition to verify(...)) so the test fails if query() starts ignoring/transforming the persistence result. ########## 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"); + } Review Comment: The canonical constructor copies the lists but doesn’t validate them first, so a null inputs will throw a NullPointerException without a clear parameter-specific message (from List.copyOf). Add Objects.requireNonNull checks for datasets/edges/columnEdges with explicit messages before copying to improve debuggability and API robustness. ########## 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); + } Review Comment: upstream/downstream are copied without an explicit null check, which can yield a generic NullPointerException from List.copyOf. Add Objects.requireNonNull(upstream, ...) and Objects.requireNonNull(downstream, ...) before copying for clearer failures and a more user-friendly API. ########## polaris-core/src/main/java/org/apache/polaris/core/lineage/LineageNode.java: ########## @@ -0,0 +1,55 @@ +/* + * 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.List; +import java.util.Objects; + +/** A node returned in a lineage query response. */ +public record LineageNode( + String id, + LineageNodeType type, + @Nullable 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(fieldMappings); + } Review Comment: fieldMappings is copied without an explicit null check, which can produce an unhelpful NullPointerException from List.copyOf. Add Objects.requireNonNull(fieldMappings, \"fieldMappings must be non-null\") before copying so callers get a clear parameter name in failures. ########## runtime/service/src/main/java/org/apache/polaris/service/lineage/DisabledLineagePersistence.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.service.lineage; + +import jakarta.enterprise.context.ApplicationScoped; +import java.time.Instant; +import java.util.List; +import org.apache.polaris.core.context.RealmContext; +import org.apache.polaris.core.lineage.LineageColumnEdge; +import org.apache.polaris.core.lineage.LineageDataset; +import org.apache.polaris.core.lineage.LineageEdge; +import org.apache.polaris.core.lineage.LineageGraph; +import org.apache.polaris.core.lineage.LineagePersistence; +import org.apache.polaris.core.lineage.LineageQueryRequest; + +/** Placeholder persistence until a concrete lineage backend is added. */ +@ApplicationScoped +public class DisabledLineagePersistence implements LineagePersistence { + private static final String MESSAGE = + "No lineage persistence implementation is configured for this deployment."; Review Comment: This error is likely to surface operationally; it would be more actionable if it pointed to the specific configuration knobs (e.g., polaris.lineage.persistence.enabled and/or polaris.lineage.persistence.type) or indicated which implementation(s) are supported. Updating the message would make failures easier to triage. ########## runtime/service/src/main/java/org/apache/polaris/service/lineage/DefaultLineageService.java: ########## @@ -0,0 +1,78 @@ +/* + * 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.service.lineage; + +import jakarta.enterprise.context.RequestScoped; +import jakarta.inject.Inject; +import java.time.Instant; +import org.apache.polaris.core.config.FeatureConfiguration; +import org.apache.polaris.core.context.CallContext; +import org.apache.polaris.core.lineage.LineageGraph; +import org.apache.polaris.core.lineage.LineageIngestRequest; +import org.apache.polaris.core.lineage.LineagePersistence; +import org.apache.polaris.core.lineage.LineageQueryRequest; +import org.apache.polaris.core.lineage.LineageService; + +@RequestScoped +public class DefaultLineageService implements LineageService { + private final CallContext callContext; + private final LineageConfiguration configuration; + private final LineagePersistence persistence; + + @Inject + public DefaultLineageService( + CallContext callContext, LineageConfiguration configuration, LineagePersistence persistence) { + this.callContext = callContext; + this.configuration = configuration; + this.persistence = persistence; + } + + @Override + public void ingest(LineageIngestRequest request) { + ensureEnabled(); + Instant lastEventAt = request.eventTime().orElseGet(Instant::now); + persistence.upsertDatasets(callContext.getRealmContext(), request.datasets()); + persistence.upsertDatasetEdges(callContext.getRealmContext(), request.edges(), lastEventAt); + persistence.upsertColumnEdges( + callContext.getRealmContext(), request.columnEdges(), lastEventAt); + } + + @Override + public LineageGraph query(LineageQueryRequest request) { + ensureEnabled(); + return persistence.loadLineage(callContext.getRealmContext(), request); + } + + private void ensureEnabled() { + if (!configuration.enabled()) { + throw new UnsupportedOperationException( + "Lineage is disabled: set polaris.lineage.enabled=true to enable it."); + } + + if (!callContext.getRealmConfig().getConfig(FeatureConfiguration.ENABLE_LINEAGE)) { + throw new UnsupportedOperationException( + "Feature not enabled: " + FeatureConfiguration.ENABLE_LINEAGE.key()); + } Review Comment: The thrown message is not very actionable because it only prints the feature key. Consider including guidance on where/how to enable the realm feature (e.g., referencing the realm feature flag name and that it must be enabled in realm config) to reduce support/debug time. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
