Copilot commented on code in PR #17789: URL: https://github.com/apache/pinot/pull/17789#discussion_r2868521182
########## pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/OfflineUpsertTableIntegrationTest.java: ########## @@ -0,0 +1,308 @@ +/** + * 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.pinot.integration.tests; + +import java.io.File; +import java.io.IOException; +import java.util.ArrayList; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import javax.annotation.Nullable; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.client.ResultSet; +import org.apache.pinot.common.utils.TarCompressionUtils; +import org.apache.pinot.segment.local.segment.creator.impl.SegmentIndexCreationDriverImpl; +import org.apache.pinot.segment.local.segment.readers.GenericRowRecordReader; +import org.apache.pinot.segment.spi.creator.SegmentGeneratorConfig; +import org.apache.pinot.spi.config.table.ColumnPartitionConfig; +import org.apache.pinot.spi.config.table.ReplicaGroupStrategyConfig; +import org.apache.pinot.spi.config.table.RoutingConfig; +import org.apache.pinot.spi.config.table.SegmentPartitionConfig; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.config.table.UpsertConfig; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.data.readers.GenericRow; +import org.apache.pinot.spi.data.readers.RecordReader; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.apache.pinot.util.TestUtils; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.testng.Assert.assertEquals; + + +/** + * Integration test for offline table upsert support. + * + * Tests that OFFLINE tables with upsert enabled correctly deduplicate records by primary key, + * keeping the latest record based on the comparison column (time column). + * + * Test data layout: + * Segment 1 (partition 0): playerId=100 (score=2000, ts=1000), playerId=101 (score=3000, ts=1000) + * Segment 2 (partition 0): playerId=100 (score=2500, ts=2000), playerId=102 (score=4000, ts=1000) + * Segment 3 (partition 0): playerId=101 (score=3500, ts=2000), playerId=102 (score=4500, ts=2000) + * + * After upsert dedup (latest by timestampInEpoch): + * playerId=100 -> score=2500 (from segment 2, ts=2000) + * playerId=101 -> score=3500 (from segment 3, ts=2000) + * playerId=102 -> score=4500 (from segment 3, ts=2000) + */ +public class OfflineUpsertTableIntegrationTest extends BaseClusterIntegrationTest { + private static final String TABLE_NAME = "offlineUpsertTest"; + private static final String OFFLINE_TABLE_NAME = TableNameBuilder.OFFLINE.tableNameWithType(TABLE_NAME); + private static final String PRIMARY_KEY_COL = "playerId"; + private static final String TIME_COL_NAME = "timestampInEpoch"; + private static final int NUM_PARTITIONS = 1; + private static final int TOTAL_RAW_RECORDS = 6; + private static final int UNIQUE_PRIMARY_KEYS = 3; + + @BeforeClass + public void setUp() + throws Exception { + TestUtils.ensureDirectoriesExistAndEmpty(_tempDir, _segmentDir, _tarDir); + + // Start the Pinot cluster + startZk(); + startController(); + startBroker(); + startServer(); + + // Create and upload schema + Schema schema = createUpsertSchema(); + addSchema(schema); + + // Create OFFLINE table config with upsert enabled + TableConfig tableConfig = createOfflineUpsertTableConfig(); + addTableConfig(tableConfig); + + // Build and upload segments with overlapping primary keys + buildAndUploadTestSegments(tableConfig, schema); + + // Wait for all documents to load + waitForAllDocsLoaded(600_000L); + } + + @AfterClass + public void tearDown() + throws IOException { + dropOfflineTable(OFFLINE_TABLE_NAME); + stopServer(); Review Comment: `dropOfflineTable(String tableName)` already appends the OFFLINE suffix internally (via `TableNameBuilder.OFFLINE.tableNameWithType(tableName)`), so passing `OFFLINE_TABLE_NAME` here will attempt to drop `..._OFFLINE_OFFLINE` and likely leave the table behind (and fail cleanup). Pass the raw table name (e.g. `TABLE_NAME`) instead, and consider removing `OFFLINE_TABLE_NAME` if it becomes unused. ########## pinot-segment-local/src/main/java/org/apache/pinot/segment/local/utils/TableConfigUtils.java: ########## @@ -782,9 +782,10 @@ static void validateUpsertAndDedupConfig(TableConfig tableConfig, Schema schema) // check both upsert and dedup are not enabled simultaneously Preconditions.checkState(!(isUpsertEnabled && isDedupEnabled), "A table can have either Upsert or Dedup enabled, but not both"); - // check table type is realtime - Preconditions.checkState(tableConfig.getTableType() == TableType.REALTIME, - "Upsert/Dedup table is for realtime table only."); + // check table type is realtime or offline + Preconditions.checkState( + tableConfig.getTableType() == TableType.REALTIME || tableConfig.getTableType() == TableType.OFFLINE, + "Upsert/Dedup table must be either realtime or offline table type."); Review Comment: With the new comparison-column fallback (configured columns → time column → segment creation time), an upsert table can now legitimately have *no* time column and an *empty* comparison-columns list. However, `validateTTLForUpsertConfig()` (called below) assumes that if comparison columns are empty it can always read `tableConfig.getValidationConfig().getTimeColumnName()` and look it up in the schema, which will throw (NPE/ISE) when the time column is not configured. Please add an explicit validation for the TTL case: if `metadataTTL`/`deletedKeysTTL` is set, require either a single comparison column or a non-null time column, and fail with a clear message instead of relying on the later schema lookup. ```suggestion "Upsert/Dedup table must be either realtime or offline table type."); // When TTL is configured for upsert, require either a single comparison column or a non-null time column. if (isUpsertEnabled && tableConfig.getUpsertConfig() != null) { UpsertConfig upsertConfig = tableConfig.getUpsertConfig(); Long metadataTTL = upsertConfig.getMetadataTTL(); Long deletedKeysTTL = upsertConfig.getDeletedKeysTTL(); boolean isTTLConfigured = (metadataTTL != null && metadataTTL > 0) || (deletedKeysTTL != null && deletedKeysTTL > 0); if (isTTLConfigured) { List<String> comparisonColumns = upsertConfig.getComparisonColumns(); boolean hasSingleComparisonColumn = comparisonColumns != null && comparisonColumns.size() == 1 && StringUtils.isNotBlank( comparisonColumns.get(0)); String timeColumnName = tableConfig.getValidationConfig() != null ? tableConfig.getValidationConfig().getTimeColumnName() : null; Preconditions.checkState(hasSingleComparisonColumn || StringUtils.isNotBlank(timeColumnName), "Upsert table with metadataTTL or deletedKeysTTL configured must specify either a single comparison " + "column or a non-null time column"); } } ``` -- 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] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
