xiangfu0 commented on code in PR #17939: URL: https://github.com/apache/pinot/pull/17939#discussion_r4214200111
########## pinot-controller/src/main/java/org/apache/pinot/controller/helix/core/ingest/InsertStatementCoordinator.java: ########## @@ -0,0 +1,1914 @@ +/** + * 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.controller.helix.core.ingest; + +import com.google.common.annotations.VisibleForTesting; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; +import java.util.EnumMap; +import java.util.HashSet; +import java.util.List; +import java.util.Map; +import java.util.Objects; +import java.util.Set; +import java.util.concurrent.ConcurrentHashMap; +import java.util.concurrent.Executors; +import java.util.concurrent.ScheduledExecutorService; +import java.util.concurrent.ThreadLocalRandom; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicLong; +import java.util.function.Consumer; +import java.util.function.Predicate; +import java.util.regex.Pattern; +import javax.annotation.Nullable; +import org.apache.pinot.common.metrics.ControllerGauge; +import org.apache.pinot.common.metrics.ControllerMeter; +import org.apache.pinot.common.metrics.ControllerMetrics; +import org.apache.pinot.controller.helix.core.PinotHelixResourceManager; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.ingest.InsertConsistencyMode; +import org.apache.pinot.spi.ingest.InsertErrorCode; +import org.apache.pinot.spi.ingest.InsertExecutor; +import org.apache.pinot.spi.ingest.InsertRequest; +import org.apache.pinot.spi.ingest.InsertResult; +import org.apache.pinot.spi.ingest.InsertStatementState; +import org.apache.pinot.spi.ingest.InsertType; +import org.apache.pinot.spi.utils.builder.TableNameBuilder; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + + +/// Central coordinator for the push-based INSERT INTO statement lifecycle. +/// +/// Manages the v1 state machine (ACCEPTED -> VISIBLE on success, ACCEPTED -> ABORTED on failure; GC'd by +/// deleting the manifest from ZK after a retention window), +/// idempotency checking, hybrid table validation, delegation to {@link InsertExecutor} backends, +/// and background cleanup of stuck statements. +/// +/// The coordinator is resilient to controller failover: on startup, it resumes cleanup from +/// the ZK-persisted state. No in-memory state is required for correctness; the ZK manifests are +/// the source of truth. +/// +/// This class is thread-safe. It is instantiated once during controller startup and shared +/// across REST API handlers. +public class InsertStatementCoordinator { + private static final Logger LOGGER = LoggerFactory.getLogger(InsertStatementCoordinator.class); + + /// Default timeout for statements stuck in ACCEPTED state (30 minutes). + public static final long DEFAULT_STATEMENT_TIMEOUT_MS = TimeUnit.MINUTES.toMillis(30); + + /// Default retention period before VISIBLE statements are moved to GC (24 hours). + public static final long DEFAULT_VISIBLE_RETENTION_MS = TimeUnit.HOURS.toMillis(24); + + /// Long retention before stuck-ACCEPTED ROW manifests are GC'd (7 days). Such manifests can occur + /// when the ROW executor's success path uploads/registers segments but the subsequent manifest + /// CAS-persist to VISIBLE fails after retries — the data is live but the state is stuck. We do + /// not abort these manifests (would falsely report failure) but we do reap them after a long + /// window so /insert/list does not accumulate stuck rows indefinitely. Operators have multiple + /// days to investigate before the manifest disappears. + public static final long ACCEPTED_ROW_RETENTION_MS = TimeUnit.DAYS.toMillis(7); + + /// Interval between cleanup sweeps (5 minutes). + private static final long CLEANUP_INTERVAL_MS = TimeUnit.MINUTES.toMillis(5); + + /// Standard Pinot segment-name character set. Requires the FIRST character to be alnum or + /// underscore/dash (NOT a dot) so adversarial inputs like ".", "..", "..." cannot survive — those + /// would otherwise be accepted by a naive `[A-Za-z0-9_.\-]+` regex and become path-traversal + /// vectors if downstream code joins segment names into deep-store paths. Also rejects consecutive + /// dots (e.g. `seg..bad`) so that adversarial relative-path fragments mid-name cannot + /// survive operating-system-specific path normalization downstream. + private static final Pattern SEGMENT_NAME_PATTERN = + Pattern.compile("[A-Za-z0-9_\\-][A-Za-z0-9_\\-]*(\\.[A-Za-z0-9_\\-]+)*"); + + /// Per-segment-name length cap for `/insert/complete`. ZK znodes have a 1MB default limit; + /// a malicious caller submitting a 1MB segment-name would otherwise wedge the manifest write. + /// 256 chars matches Pinot's typical segment-name budget. + private static final int MAX_SEGMENT_NAME_LENGTH = 256; + + /// Per-completion segment-name list cap. 10K is far above any realistic single-batch size and + /// keeps the manifest's segmentNames within ZK's znode size budget even with maximum-length names. + private static final int MAX_SEGMENT_NAMES_PER_COMPLETION = 10_000; + + private final PinotHelixResourceManager _helixResourceManager; + private final InsertStatementStore _statementStore; + private final ControllerMetrics _controllerMetrics; + /// Mutable until {@link #start()} freezes the registration into {@link #_executorSnapshot}. After + /// start, no thread reads this map — all lookups go through the immutable snapshot. EnumMap is + /// used because {@link InsertType} is a small fixed enum; the EnumMap variant is faster than + /// HashMap (array-backed) and better self-documents the closed key domain. + private final EnumMap<InsertType, InsertExecutor> _executors; + /// Immutable snapshot of {@link #_executors} published at {@link #start()}. The `volatile` + /// write in `start()` establishes a happens-before with every subsequent volatile read on + /// the request paths, removing the need to reason about the underlying map's per-key visibility. + /// `null` until `start()` runs. + private volatile Map<InsertType, InsertExecutor> _executorSnapshot; + private final long _statementTimeoutMs; + private final long _visibleRetentionMs; + + /// Tables that are known to have insert statements. Populated as statements are submitted + /// and during cleanup sweeps. This set is an optimization to avoid scanning all tables; the + /// ZK state is always the source of truth. + private final Set<String> _tablesWithStatements = ConcurrentHashMap.newKeySet(); + + /// `true` once {@link #start()} has been called. Controls whether the REST layer will + /// accept traffic and whether {@link #stop()} needs to shut down the scheduler. When the feature + /// flag is off, the coordinator is still constructed (so Jersey binding resolves) but this stays + /// `false`. + private volatile boolean _started = false; + private volatile ScheduledExecutorService _cleanupSchedulerInstance; + /// Counts in-flight `submitInsert` calls (after `_started` check, before the executor + /// call returns). `stop()` awaits this counter draining to zero before returning, so a + /// long-running `executor.execute()` cannot land a ZK write after the coordinator is + /// officially stopped — which would break failover semantics if a new leader has begun running. + private final AtomicLong _inflightSubmitCount = new AtomicLong(0); + + /// Counts in-flight {@link #cleanupAllTables} sweep iterations. The sweep mutates ZK on every + /// state transition (CAS-flips, manifest deletes) and can run for tens of seconds with thousands + /// of manifests. `stop()` awaits this counter draining alongside `_inflightSubmitCount` + /// so a sweep cannot land a write after stop() returns and a new leader on a different controller + /// begins running. + private final AtomicLong _inflightSweepCount = new AtomicLong(0); + + /// Atomic counter of statements currently in a non-terminal state. In v1 the only non-terminal + /// state is {@link InsertStatementState#ACCEPTED} — PREPARED/COMMITTED are reserved for v2's + /// two-phase commit. The counter is seeded once at {@link #start()} via + /// {@link #computeActiveStatementCountFromZk()}, adjusted incrementally by transition sites + /// (submit, abort, complete, cleanup-sweep), and periodically reseeded by the cleanup sweep + /// (every {@link #ACTIVE_COUNT_RESEED_EVERY_N_SWEEPS} passes) to reconcile any drift from rare + /// race-induced double-counts (e.g., the EXECUTOR_ERROR_BUT_DURABLE path can decrement + /// speculatively when the manifest re-read raced with a writer). Constant-time read; converges + /// to ZK truth on every controller restart and on each periodic reseed. + private final AtomicLong _activeStatementCount = new AtomicLong(0); + + /// Counter of cleanup sweeps since {@link #start()}; used to drive periodic reseed of + /// {@link #_activeStatementCount} from ZK without doing it on every 5-minute sweep. + private final AtomicLong _cleanupSweepCount = new AtomicLong(0); + + /// Reseed the active-statement counter from ZK every Nth cleanup sweep. At a 5-minute sweep + /// interval, 12 sweeps = 1 hour between reseeds. Reseeding every sweep doubles the ZK list cost + /// for clusters with thousands of manifests; bounding drift to 1 hour is acceptable for an + /// advisory gauge. The incremental counter is the authoritative source between reseeds. + private static final int ACTIVE_COUNT_RESEED_EVERY_N_SWEEPS = 12; + + public InsertStatementCoordinator(PinotHelixResourceManager helixResourceManager, + InsertStatementStore statementStore, ControllerMetrics controllerMetrics) { + this(helixResourceManager, statementStore, controllerMetrics, + DEFAULT_STATEMENT_TIMEOUT_MS, DEFAULT_VISIBLE_RETENTION_MS); + } + + public InsertStatementCoordinator(PinotHelixResourceManager helixResourceManager, + InsertStatementStore statementStore, ControllerMetrics controllerMetrics, + long statementTimeoutMs, long visibleRetentionMs) { + _helixResourceManager = helixResourceManager; + _statementStore = statementStore; + _controllerMetrics = controllerMetrics; + _executors = new EnumMap<>(InsertType.class); + _statementTimeoutMs = statementTimeoutMs; + _visibleRetentionMs = visibleRetentionMs; + /// Do NOT create the scheduler thread here — when the feature flag is disabled we never call + /// start() and the thread would be leaked. Deferred to start(). + } + + /// Starts the background cleanup task for stuck statements. Creates the scheduler thread lazily + /// so that when the feature flag is off and `start()` is never called, no thread is + /// allocated. + /// On startup (including after controller failover), the first sweep picks up any + /// stuck statements from ZK. + public synchronized void start() { + if (_started) { + return; + } + /// Publish the executor snapshot, seed the active-counter, and flip _started=true BEFORE arming + /// the scheduler. The cleanup sweep's first iteration must observe a fully-initialized + /// coordinator; arming the scheduler last ensures that even the rare interleaving where the + /// scheduled task fires immediately (theoretical on busy CPU) sees the published state. + _executorSnapshot = Collections.unmodifiableMap(new EnumMap<>(_executors)); + /// Seed the active-statement counter from ZK so that recovery after a previous leader's crash + /// starts from the correct value. We discover tables lazily — _tablesWithStatements is empty + /// here; the first cleanup sweep will populate it and reseed. + _activeStatementCount.set(computeActiveStatementCountFromZk()); + _started = true; + + ScheduledExecutorService scheduler = Executors.newSingleThreadScheduledExecutor(r -> { + Thread t = new Thread(r, "insert-statement-cleanup"); + t.setDaemon(true); + return t; + }); + try { + scheduler.scheduleWithFixedDelay(this::cleanupAllTables, CLEANUP_INTERVAL_MS, CLEANUP_INTERVAL_MS, + TimeUnit.MILLISECONDS); + } catch (RuntimeException e) { + /// Don't leak the executor if scheduling fails — roll back _started and surface the error. + _started = false; + _executorSnapshot = null; + scheduler.shutdownNow(); + throw e; + } + _cleanupSchedulerInstance = scheduler; + LOGGER.info("InsertStatementCoordinator started with statementTimeout={}ms, visibleRetention={}ms", + _statementTimeoutMs, _visibleRetentionMs); + } + + /// @return `true` once {@link #start()} has been called. REST handlers use this to fail + /// closed with HTTP 503 when the coordinator is idle (feature flag off). + public boolean isStarted() { + return _started; + } + + /// Stops the coordinator and shuts down the cleanup scheduler. Order matters: flip + /// `_started=false` BEFORE the shutdown so any new sweep iteration bails immediately, then + /// `shutdownNow()` interrupts in-flight work, then await termination so a sweep that was + /// mid-write to ZK gets time to finish before `stop()` returns. Without the await, an + /// in-flight ZK write can land after the controller is officially stopped — and a new leader on + /// a different controller observes the dying leader's mutation, breaking failover semantics. + public synchronized void stop() { + _started = false; + /// Await any in-flight submitInsert calls. New calls bail at the _started check above; existing + /// ones (already past the check, possibly mid-executor.execute()) may take longer than the + /// sweep scheduler's timeout. Polling once per 100ms is fine for stop() — it runs once per + /// controller lifetime. Cap at 30s to avoid blocking shutdown indefinitely if a request is + /// truly wedged (e.g., a Minion call that never returns); operators see the WARN. + long awaitDeadlineMs = System.currentTimeMillis() + 30_000; + while ((_inflightSubmitCount.get() > 0 || _inflightSweepCount.get() > 0) + && System.currentTimeMillis() < awaitDeadlineMs) { + try { + Thread.sleep(100); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + break; + } + } + long stillInflightSubmits = _inflightSubmitCount.get(); + long stillInflightSweeps = _inflightSweepCount.get(); + if (stillInflightSubmits > 0 || stillInflightSweeps > 0) { + LOGGER.warn("InsertStatementCoordinator stop(): {} in-flight submitInsert calls and {} " + + "in-flight sweep iterations did not drain within 30s; ZK writes from those may " + + "land after stop() returns", stillInflightSubmits, stillInflightSweeps); + } + if (_cleanupSchedulerInstance != null) { + _cleanupSchedulerInstance.shutdownNow(); + try { + if (!_cleanupSchedulerInstance.awaitTermination(10, TimeUnit.SECONDS)) { + LOGGER.warn("InsertStatementCoordinator cleanup scheduler did not terminate within 10s; " + + "in-flight ZK writes may complete after stop() returns"); + } + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + LOGGER.warn("Interrupted while waiting for cleanup scheduler shutdown"); + } + _cleanupSchedulerInstance = null; + } + /// Defense-in-depth: drop the frozen executor snapshot symmetrically with the start() + /// rollback path. The _started flag is the load-bearing guard for stopping new submissions + /// and sweep iterations; this null-out only ensures the snapshot is not held longer than the + /// coordinator's lifecycle (lookupExecutor falls back to the live _executors map, so this is + /// not a hard fence — it's hygiene). + _executorSnapshot = null; + LOGGER.info("InsertStatementCoordinator stopped"); + } + + /// Registers an {@link InsertExecutor} for a specific insert type. Must be called BEFORE + /// {@link #start()} — the cleanup sweep and submitInsert paths read from `_executors` + /// without holding a lock, on the assumption that the registration set is frozen at start time. + /// + /// **v1 limitation:** `InsertType.FILE` requires the in-tree {@link FileInsertExecutor} + /// concrete class. The coordinator hardcodes `instanceof FileInsertExecutor` checks at + /// several lifecycle sites (task-name caching, Minion task polling, completion). A third-party + /// plugin that implements {@link InsertExecutor} alone and registers as `FILE` would + /// silently bypass those hooks, leaving manifests stuck in `ACCEPTED`. We fail fast here + /// rather than silently accept a half-functional registration. v2 will lift FILE-specific hooks + /// into the SPI as default methods so third-party implementations can opt in. + /// + /// @throws IllegalStateException if called after {@link #start()} has flipped `_started` + /// @throws UnsupportedOperationException if the executor for `InsertType.FILE` is not a + /// subclass of the in-tree {@link FileInsertExecutor} + public synchronized void registerExecutor(String executorType, InsertExecutor executor) { + if (_started) { + throw new IllegalStateException("registerExecutor must be called before start(); coordinator is already started. " + + "Late registration would race with the cleanup sweep and submitInsert."); + } + InsertType type; + try { + type = InsertType.valueOf(executorType); + } catch (IllegalArgumentException e) { + throw new IllegalArgumentException("Unknown InsertExecutor type: '" + executorType + + "'. Supported values: " + Arrays.toString(InsertType.values()), e); + } + if (type == InsertType.FILE && !(executor instanceof FileInsertExecutor)) { + throw new UnsupportedOperationException("InsertType.FILE requires a FileInsertExecutor (or subclass) in v1. " + + "Pluggable third-party FILE executors are not supported until v2 lifts FILE-specific hooks " + + "(getScheduledTaskName, taskCompleted, prepareCompletionResult) into the InsertExecutor SPI. " + + "Got executor of class: " + executor.getClass().getName()); + } + _executors.put(type, executor); + LOGGER.info("Registered InsertExecutor for type={}", type); + } + + /// Reads an executor from the immutable post-{@link #start()} snapshot. Falls back to the + /// still-mutable {@link #_executors} map only when `_started=false` (i.e., from tests that + /// register and read without starting). Production code paths always go through the snapshot. + @Nullable + private InsertExecutor lookupExecutor(InsertType type) { + Map<InsertType, InsertExecutor> snapshot = _executorSnapshot; + if (snapshot != null) { + return snapshot.get(type); + } + return _executors.get(type); + } + + @Nullable + private InsertExecutor lookupExecutor(String executorTypeName) { + try { + return lookupExecutor(InsertType.valueOf(executorTypeName)); + } catch (IllegalArgumentException e) { + return null; + } + } + + /// Submits an INSERT INTO request for execution. + /// + /// Performs idempotency checks, hybrid table validation, creates a manifest, + /// and delegates to the appropriate executor. + /// + /// @param request the insert request + /// @return the result reflecting the initial state of the statement + public InsertResult submitInsert(InsertRequest request) { + /// Increment the in-flight counter BEFORE the _started check. This ordering is load-bearing + /// for stop()'s drain: stop() flips _started=false then awaits _inflightSubmitCount==0. If we + /// checked _started first and a concurrent stop() raced between the check and the increment, + /// stop() would return while we were still about to do work — ZK writes could land after the + /// coordinator is officially stopped, breaking failover semantics. By incrementing first then + /// re-checking _started, any request that gets past the check is guaranteed observable to + /// stop()'s drain loop. + _inflightSubmitCount.incrementAndGet(); + try { + if (!_started) { + /// Feature flag is off OR coordinator is mid-startup/shutdown. Fail fast for in-process + /// callers (e.g., the local SqlQueryExecutor wrapper) — REST callers see HTTP 503 via + /// InsertStatementResource.checkEnabled(). Both paths must reject submissions consistently. + return rejectResult(request.getStatementId(), InsertErrorCode.COORDINATOR_NOT_READY, + "InsertStatementCoordinator is not started (feature disabled or controller starting/stopping)"); + } + return submitInsertInternal(request); + } finally { + _inflightSubmitCount.decrementAndGet(); + } + } + + private InsertResult submitInsertInternal(InsertRequest request) { Review Comment: I narrowed this PR to synchronous ROW inserts with a create-only requestId reservation, retained until table deletion even after failed or uncertain execution. FILE coordination, cancellation, sweep/GC and rebind are removed from this PR. The final focused unit and two-controller cluster runs passed at 14ddef8878. ########## pinot-integration-tests/src/test/java/org/apache/pinot/integration/tests/InsertIntoValuesClusterIntegrationTest.java: ########## @@ -0,0 +1,404 @@ +/** + * 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 com.fasterxml.jackson.databind.JsonNode; +import java.io.IOException; +import java.net.URLEncoder; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.commons.io.FileUtils; +import org.apache.pinot.common.exception.HttpErrorStatusException; +import org.apache.pinot.controller.ControllerConf; +import org.apache.pinot.spi.config.table.TableConfig; +import org.apache.pinot.spi.config.table.TableType; +import org.apache.pinot.spi.data.FieldSpec; +import org.apache.pinot.spi.data.Schema; +import org.apache.pinot.spi.ingest.InsertConsistencyMode; +import org.apache.pinot.spi.ingest.InsertType; +import org.apache.pinot.spi.utils.JsonUtils; +import org.apache.pinot.spi.utils.builder.TableConfigBuilder; +import org.apache.pinot.util.TestUtils; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; +import org.testng.annotations.AfterClass; +import org.testng.annotations.BeforeClass; +import org.testng.annotations.Test; + +import static org.testng.Assert.*; + + +/// Integration tests for push-based INSERT INTO functionality. +/// +/// Tests the controller coordinator REST API for statement lifecycle management, +/// idempotency, hybrid table validation, abort, and list operations. +/// +/// The ROW executor is registered with the coordinator at controller startup, +/// so submitted inserts are expected to succeed (state = VISIBLE). +/// +/// **Base class choice:** this test extends {@link BaseClusterIntegrationTest} rather than +/// {@link org.apache.pinot.integration.tests.custom.CustomDataQueryClusterIntegrationTest} +/// for three concrete reasons: +/// - **Per-test table topology** — each test method creates its own OFFLINE and/or REALTIME +/// table inside the test body and tears it down at the end. This is required because tests +/// exercise hybrid table validation (which needs both _OFFLINE and _REALTIME variants present), +/// table-not-found rejection (needs a table to NOT exist), and idempotency across distinct +/// tables. The `CustomDataQueryClusterIntegrationTest` pattern of one shared schema/table +/// per test class cannot express these topologies. +/// - **Coordinator state isolation** — each test asserts on coordinator-internal state +/// (active-statement counter, request-id reservation tombstones, manifest GC). Sharing tables +/// across tests would let one test's manifest leak into another's listStatements assertions. +/// - **Controller config override** — {@link #overrideControllerConf} sets +/// `controller.insert.enabled=true`. `BaseClusterIntegrationTest` runs setUp/tearDown +/// per class so the flag binding is stable; cluster restart per method is not needed. +public class InsertIntoValuesClusterIntegrationTest extends BaseClusterIntegrationTest { + private static final Logger LOGGER = + LoggerFactory.getLogger(InsertIntoValuesClusterIntegrationTest.class); + + @Override + protected void overrideControllerConf(Map<String, Object> properties) { + /// Enable the push-based INSERT INTO feature flag. The flag defaults to false in production + /// so operators must opt in; integration tests need it on to exercise the coordinator. + properties.put(ControllerConf.INSERT_ENABLED, true); + } + + @BeforeClass + public void setUp() + throws Exception { + TestUtils.ensureDirectoriesExistAndEmpty(_tempDir); + startZk(); + startController(); + startBroker(); + startServer(); + } + + @AfterClass + public void tearDown() { + try { + stopServer(); + stopBroker(); + stopController(); + stopZk(); + } finally { + FileUtils.deleteQuietly(_tempDir); + } + } + + private void createOfflineTable(String tableName) + throws Exception { + Schema schema = new Schema.SchemaBuilder() + .setSchemaName(tableName) + .addSingleValueDimension("id", FieldSpec.DataType.INT) + .addSingleValueDimension("name", FieldSpec.DataType.STRING) + .addMetric("score", FieldSpec.DataType.FLOAT) + .build(); + addSchema(schema); + + TableConfig tableConfig = new TableConfigBuilder(TableType.OFFLINE) + .setTableName(tableName) + .build(); + sendPostRequest( + _controllerRequestURLBuilder.forTableCreate(), + tableConfig.toString(), + BasicAuthTestUtils.AUTH_HEADER); + } + + private String buildInsertRequestJson(String tableName, TableType tableType, String requestId, + String payloadHash) + throws Exception { + Map<String, Object> request = new HashMap<>(); + request.put("tableName", tableName); + request.put("insertType", InsertType.ROW.name()); + request.put("consistencyMode", InsertConsistencyMode.WAIT_FOR_ACCEPT.name()); + + if (tableType != null) { + request.put("tableType", tableType.name()); + } + if (requestId != null) { + request.put("requestId", requestId); + } + if (payloadHash != null) { + request.put("payloadHash", payloadHash); + } + + /// Minimal rows + java.util.List<Map<String, Object>> rowList = new java.util.ArrayList<>(); + Map<String, Object> row = new HashMap<>(); + Map<String, Object> fieldMap = new HashMap<>(); + fieldMap.put("id", 1); + fieldMap.put("name", "test"); + fieldMap.put("score", 90.0); + row.put("fieldToValueMap", fieldMap); + rowList.add(row); + request.put("rows", rowList); + + return JsonUtils.objectToString(request); + } + + private JsonNode postInsertRequest(String payload) + throws Exception { + /// The /insert/execute endpoint requires ?tableName=<table> as a query param to bind the + /// table-scoped @Authorize check. Extract the value from the payload so callers don't need to + /// pass it twice. + String tableName = JsonUtils.stringToJsonNode(payload).get("tableName").asText(); + String url = getControllerBaseApiUrl() + "/insert/execute?tableName=" + + URLEncoder.encode(tableName, StandardCharsets.UTF_8); + String response = sendPostRequest(url, payload, + Collections.singletonMap("accept", "application/json")); + return JsonUtils.stringToJsonNode(response); + } + + private JsonNode getInsertStatus(String statementId, String tableNameWithType) + throws Exception { + String url = getControllerBaseApiUrl() + "/insert/status/" + statementId + + "?tableName=" + tableNameWithType; + String response = sendGetRequest(url); + return JsonUtils.stringToJsonNode(response); + } + + private JsonNode postInsertAbort(String statementId, String tableNameWithType) + throws Exception { + String url = getControllerBaseApiUrl() + "/insert/abort/" + statementId + + "?tableName=" + tableNameWithType; + String response = sendPostRequest(url, null, + Collections.singletonMap("accept", "application/json")); + return JsonUtils.stringToJsonNode(response); + } + + private JsonNode getInsertList(String tableNameWithType) + throws Exception { + String url = getControllerBaseApiUrl() + "/insert/list?tableName=" + tableNameWithType; + String response = sendGetRequest(url); + return JsonUtils.stringToJsonNode(response); + } + + /// ---- Test: Submit insert and verify coordinator processes it ---- + + @Test + public void testInsertIntoOfflineTableReturnsResult() + throws Exception { + String tableName = "insertOfflineResult"; + createOfflineTable(tableName); + + String payload = buildInsertRequestJson(tableName, null, null, null); + + JsonNode result = postInsertRequest(payload); + LOGGER.info("Insert result: {}", result); + + assertNotNull(result.get("statementId"), "statementId should be present"); + String statementId = result.get("statementId").asText(); + assertFalse(statementId.isEmpty(), "statementId should not be empty"); + + /// With ROW executor registered, the insert should reach VISIBLE synchronously. + String state = result.get("state").asText(); + assertEquals(state, "VISIBLE", "ROW executor should return VISIBLE synchronously, got: " + state); + } + + /// ---- Test: Explicit OFFLINE type targeting works for single-type table ---- + + @Test + public void testInsertWithExplicitOfflineType() + throws Exception { + String tableName = "insertExplicitOffline"; + createOfflineTable(tableName); + + /// Submit with explicit tableType=OFFLINE on a table that only has OFFLINE + String payload = buildInsertRequestJson(tableName, TableType.OFFLINE, null, null); + JsonNode result = postInsertRequest(payload); + LOGGER.info("Explicit OFFLINE type result: {}", result); + /// Assert the request resolves and reaches VISIBLE — pins the explicit-type resolution path. + assertEquals(result.get("state").asText(), "VISIBLE", + "Explicit OFFLINE type on an OFFLINE-only table should resolve and produce VISIBLE; got: " + result); + } + + /// ---- Test: Table does not exist ---- + + @Test + public void testInsertNonExistentTable() + throws Exception { + String payload = buildInsertRequestJson("nonExistentTable99", null, null, null); + + JsonNode result = postInsertRequest(payload); + LOGGER.info("Non-existent table result: {}", result); + assertEquals(result.get("state").asText(), "REJECTED"); + assertEquals(result.get("errorCode").asText(), "TABLE_RESOLUTION_ERROR"); + } + + /// ---- Test: Insert with wrong explicit table type ---- + + @Test + public void testInsertWithWrongTableType() + throws Exception { + String tableName = "insertWrongType"; + createOfflineTable(tableName); + + /// Try to insert with REALTIME type into an OFFLINE-only table + String payload = buildInsertRequestJson(tableName, TableType.REALTIME, null, null); + JsonNode result = postInsertRequest(payload); + LOGGER.info("Wrong type result: {}", result); + assertEquals(result.get("state").asText(), "REJECTED"); + assertEquals(result.get("errorCode").asText(), "TABLE_RESOLUTION_ERROR"); + } + + /// ---- Test: Insert with explicit table type suffix in name ---- + + @Test + public void testInsertWithExplicitTableTypeSuffix() + throws Exception { + String tableName = "insertSuffix"; + createOfflineTable(tableName); + + /// Use explicit _OFFLINE suffix in table name + String payload = buildInsertRequestJson(tableName + "_OFFLINE", null, null, null); + JsonNode result = postInsertRequest(payload); + LOGGER.info("Explicit suffix result: {}", result); + String state = result.get("state").asText(); + assertEquals(state, "VISIBLE", "ROW executor should return VISIBLE synchronously, got: " + state); + } + + /// ---- Test: Idempotency — same requestId + payloadHash should return the + /// same statementId and not re-execute ---- + + @Test + public void testInsertIdempotency() + throws Exception { + String tableName = "insertIdemp"; + createOfflineTable(tableName); + + String requestId = "idemp-001"; + String payloadHash = "hash-abc"; + + String payload = buildInsertRequestJson(tableName, null, requestId, payloadHash); + + /// First submit — succeeds with executor registered + JsonNode result1 = postInsertRequest(payload); + LOGGER.info("First submit: {}", result1); + String state1 = result1.get("state").asText(); + assertFalse("REJECTED".equals(state1) && result1.has("errorCode") + && "NO_EXECUTOR".equals(result1.get("errorCode").asText()), + "Executor should be registered; should not get NO_EXECUTOR"); + String statementId1 = result1.get("statementId").asText(); + + /// Second submit with same requestId + payloadHash — should be idempotent + JsonNode result2 = postInsertRequest(payload); + LOGGER.info("Second submit: {}", result2); + assertEquals(result2.get("statementId").asText(), statementId1, + "Idempotent retry should return same statementId"); + } + + /// ---- Test: Status API returns NOT_FOUND for unknown statement ---- + + @Test + public void testInsertStatusNotFound() + throws Exception { + String tableName = "insertStatusNotFound"; + createOfflineTable(tableName); + + /// The resource maps any state=REJECTED coordinator response (including NOT_FOUND) to HTTP 404 + /// to honor the contract that REJECTED has no manifest persisted. ControllerTest.sendGetRequest + /// wraps HttpErrorStatusException in an IOException on 4xx, so we catch IOException and unwrap + /// to check the status code and error message. + try { + getInsertStatus("unknown-stmt-id", tableName + "_OFFLINE"); + fail("Expected HTTP 404 for unknown statementId"); + } catch (IOException e) { + Throwable cause = e.getCause(); + assertTrue(cause instanceof HttpErrorStatusException, + "Expected cause to be HttpErrorStatusException, got: " + cause); + HttpErrorStatusException httpEx = (HttpErrorStatusException) cause; + assertEquals(httpEx.getStatusCode(), 404); + assertTrue(httpEx.getMessage().contains("errorCode=NOT_FOUND"), + "Expected message to mention errorCode=NOT_FOUND, got: " + httpEx.getMessage()); + } + } + + /// ---- Test: Abort returns NOT_FOUND for unknown statement ---- + + @Test + public void testInsertAbortNotFound() + throws Exception { + String tableName = "insertAbortNotFound"; + createOfflineTable(tableName); + + JsonNode abortResult = postInsertAbort("unknown-stmt-id", tableName + "_OFFLINE"); + LOGGER.info("Abort not found: {}", abortResult); + assertEquals(abortResult.get("errorCode").asText(), "NOT_FOUND"); + } + + /// ---- Test: List returns empty array for table with no statements ---- + + @Test + public void testInsertListEmpty() + throws Exception { + String tableName = "insertListEmpty"; + createOfflineTable(tableName); + + JsonNode list = getInsertList(tableName + "_OFFLINE"); + LOGGER.info("Empty list: {}", list); + assertTrue(list.isArray(), "List should return an array"); + assertEquals(list.size(), 0, "Should have 0 statements"); + } + + /// ---- Test: SQL parsing of INSERT INTO VALUES through broker ---- + + @Test + public void testInsertIntoValuesSqlParsing() Review Comment: The ROW suite passed all 14 real-cluster tests: REALTIME uploaded/consuming coexistence, typed BYTES/BIG_DECIMAL/negative values, hybrid explicit-type selection, partitioned full upsert and matching server assignments. It also verifies two-controller requestId races, restart/retry retention, table deletion/recreation, SQL caller authorization and representation-sensitive retry conflicts. The 97 focused unit cases passed too; the new FILE API/executor was removed from this PR, so its integration coverage belongs with that follow-up. -- 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]
