Re: [PR] Add push-based INSERT INTO support [pinot]

["Copilot (via GitHub)" <gi...@apache.org>]

Copilot commented on code in PR #17939:
URL: https://github.com/apache/pinot/pull/17939#discussion_r3027144539


##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/LocalShardLog.java:
##########
@@ -0,0 +1,285 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.pinot.spi.ingest.ShardLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * File-based append-only implementation of {@link ShardLog}.
+ *
+ * <p>Data is written as length-prefixed entries to a log file:
+ * <pre>
+ *   [4 bytes: entry length][N bytes: entry data]
+ * </pre>
+ *
+ * <p>The offset returned by {@link #append(byte[])} is the byte offset in the log file where the
+ * entry begins. Statement metadata (prepared/committed/aborted state and offset ranges) is tracked
+ * in a separate metadata file.
+ *
+ * <p>This implementation is thread-safe for concurrent appends via a read-write lock.
+ * Prepare/commit/abort operations for a given statement are expected to be serialized externally.
+ */
+public class LocalShardLog implements ShardLog {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(LocalShardLog.class);
+  private static final String LOG_FILE_NAME = "data.log";
+  private static final String META_FILE_NAME = "meta.log";
+
+  private final File _logDir;
+  private final File _logFile;
+  private final File _metaFile;
+  private final ReadWriteLock _lock = new ReentrantReadWriteLock();
+  private final AtomicLong _writeOffset = new AtomicLong(0);
+  private final Map<String, StatementMeta> _statementMetas = new ConcurrentHashMap<>();
+
+  /**
+   * Creates a new local shard log under the given directory.
+   *
+   * @param logDir the directory for this shard log's files
+   * @throws IOException if the directory cannot be created or the log file cannot be initialized
+   */
+  public LocalShardLog(File logDir)
+      throws IOException {
+    _logDir = logDir;
+    if (!_logDir.exists() && !_logDir.mkdirs()) {
+      throw new IOException("Failed to create log directory: " + _logDir);
+    }
+    _logFile = new File(_logDir, LOG_FILE_NAME);
+    _metaFile = new File(_logDir, META_FILE_NAME);
+
+    if (_logFile.exists()) {
+      _writeOffset.set(_logFile.length());
+    }
+
+    // Recover statement metadata from the meta file
+    recoverMetadata();
+  }
+
+  @Override
+  public long append(byte[] data) {
+    _lock.writeLock().lock();
+    try {
+      long offset = _writeOffset.get();
+      try (FileOutputStream fos = new FileOutputStream(_logFile, true);
+           FileChannel channel = fos.getChannel()) {
+        ByteBuffer header = ByteBuffer.allocate(4);
+        header.putInt(data.length);
+        header.flip();
+        channel.write(header);
+        channel.write(ByteBuffer.wrap(data));
+        channel.force(true);
+      } catch (IOException e) {
+        throw new RuntimeException("Failed to append to shard log: " + _logFile, e);
+      }
+      _writeOffset.addAndGet(4 + data.length);
+      return offset;
+    } finally {
+      _lock.writeLock().unlock();
+    }
+  }
+
+  @Override
+  public void prepare(String statementId, long startOffset, long endOffset) {
+    StatementMeta meta = new StatementMeta(statementId, startOffset, endOffset, StatementStatus.PREPARED);
+    _statementMetas.put(statementId, meta);
+    persistMetaEntry(statementId, startOffset, endOffset, StatementStatus.PREPARED);
+  }
+
+  @Override
+  public void commit(String statementId) {
+    StatementMeta existing = _statementMetas.get(statementId);
+    if (existing == null) {
+      LOGGER.warn("Attempted to commit unknown statement: {}", statementId);
+      return;
+    }
+    StatementMeta committed = new StatementMeta(
+        statementId, existing._startOffset, existing._endOffset, StatementStatus.COMMITTED);
+    _statementMetas.put(statementId, committed);
+    persistMetaEntry(statementId, existing._startOffset, existing._endOffset, StatementStatus.COMMITTED);
+  }
+
+  @Override
+  public void abort(String statementId) {
+    StatementMeta existing = _statementMetas.get(statementId);
+    long startOffset = existing != null ? existing._startOffset : -1;
+    long endOffset = existing != null ? existing._endOffset : -1;
+    StatementMeta aborted = new StatementMeta(statementId, startOffset, endOffset, StatementStatus.ABORTED);
+    _statementMetas.put(statementId, aborted);
+    persistMetaEntry(statementId, startOffset, endOffset, StatementStatus.ABORTED);
+  }
+
+  @Override
+  public Iterator<byte[]> read(long fromOffset) {
+    return new LogIterator(_logFile, fromOffset, _writeOffset.get());
+  }
+
+  /**
+   * Returns the current write offset (end of log).
+   */
+  public long getWriteOffset() {
+    return _writeOffset.get();
+  }
+
+  /**
+   * Returns the metadata for the given statement, or {@code null} if not found.
+   */
+  public StatementMeta getStatementMeta(String statementId) {
+    return _statementMetas.get(statementId);
+  }
+
+  private void persistMetaEntry(String statementId, long startOffset, long endOffset, StatementStatus status) {
+    try (FileOutputStream fos = new FileOutputStream(_metaFile, true);
+         FileChannel channel = fos.getChannel()) {
+      byte[] line = (statementId + "," + startOffset + "," + endOffset + "," + status.name() + "\n")
+          .getBytes(java.nio.charset.StandardCharsets.UTF_8);
+      channel.write(ByteBuffer.wrap(line));
+      channel.force(true);
+    } catch (IOException e) {
+      String message = "Failed to persist metadata for statement: " + statementId;
+      LOGGER.error(message, e);
+      throw new IllegalStateException(message, e);

Review Comment:
   `persistMetaEntry()` appends to the shared `_metaFile` without any synchronization. Since `prepare/commit/abort` can be invoked concurrently for different statements, concurrent appends can interleave and corrupt `meta.log`, breaking recovery. Consider serializing metadata writes (e.g., synchronize on a lock / use the existing `_lock`) or using a single dedicated writer/queue.
   ```suggestion
       _lock.writeLock().lock();
       try {
         try (FileOutputStream fos = new FileOutputStream(_metaFile, true);
              FileChannel channel = fos.getChannel()) {
           byte[] line = (statementId + "," + startOffset + "," + endOffset + "," + status.name() + "\n")
               .getBytes(java.nio.charset.StandardCharsets.UTF_8);
           channel.write(ByteBuffer.wrap(line));
           channel.force(true);
         } catch (IOException e) {
           String message = "Failed to persist metadata for statement: " + statementId;
           LOGGER.error(message, e);
           throw new IllegalStateException(message, e);
         }
       } finally {
         _lock.writeLock().unlock();
   ```



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/LocalShardLog.java:
##########
@@ -0,0 +1,285 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.io.RandomAccessFile;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.util.Iterator;
+import java.util.Map;
+import java.util.NoSuchElementException;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicLong;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
+import org.apache.pinot.spi.ingest.ShardLog;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * File-based append-only implementation of {@link ShardLog}.
+ *
+ * <p>Data is written as length-prefixed entries to a log file:
+ * <pre>
+ *   [4 bytes: entry length][N bytes: entry data]
+ * </pre>
+ *
+ * <p>The offset returned by {@link #append(byte[])} is the byte offset in the log file where the
+ * entry begins. Statement metadata (prepared/committed/aborted state and offset ranges) is tracked
+ * in a separate metadata file.
+ *
+ * <p>This implementation is thread-safe for concurrent appends via a read-write lock.
+ * Prepare/commit/abort operations for a given statement are expected to be serialized externally.
+ */
+public class LocalShardLog implements ShardLog {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(LocalShardLog.class);
+  private static final String LOG_FILE_NAME = "data.log";
+  private static final String META_FILE_NAME = "meta.log";
+
+  private final File _logDir;
+  private final File _logFile;
+  private final File _metaFile;
+  private final ReadWriteLock _lock = new ReentrantReadWriteLock();
+  private final AtomicLong _writeOffset = new AtomicLong(0);
+  private final Map<String, StatementMeta> _statementMetas = new ConcurrentHashMap<>();
+
+  /**
+   * Creates a new local shard log under the given directory.
+   *
+   * @param logDir the directory for this shard log's files
+   * @throws IOException if the directory cannot be created or the log file cannot be initialized
+   */
+  public LocalShardLog(File logDir)
+      throws IOException {
+    _logDir = logDir;
+    if (!_logDir.exists() && !_logDir.mkdirs()) {
+      throw new IOException("Failed to create log directory: " + _logDir);
+    }
+    _logFile = new File(_logDir, LOG_FILE_NAME);
+    _metaFile = new File(_logDir, META_FILE_NAME);
+
+    if (_logFile.exists()) {
+      _writeOffset.set(_logFile.length());
+    }
+
+    // Recover statement metadata from the meta file
+    recoverMetadata();
+  }
+
+  @Override
+  public long append(byte[] data) {
+    _lock.writeLock().lock();
+    try {
+      long offset = _writeOffset.get();
+      try (FileOutputStream fos = new FileOutputStream(_logFile, true);
+           FileChannel channel = fos.getChannel()) {
+        ByteBuffer header = ByteBuffer.allocate(4);
+        header.putInt(data.length);
+        header.flip();
+        channel.write(header);
+        channel.write(ByteBuffer.wrap(data));
+        channel.force(true);
+      } catch (IOException e) {
+        throw new RuntimeException("Failed to append to shard log: " + _logFile, e);
+      }
+      _writeOffset.addAndGet(4 + data.length);
+      return offset;
+    } finally {
+      _lock.writeLock().unlock();
+    }
+  }
+
+  @Override
+  public void prepare(String statementId, long startOffset, long endOffset) {
+    StatementMeta meta = new StatementMeta(statementId, startOffset, endOffset, StatementStatus.PREPARED);
+    _statementMetas.put(statementId, meta);
+    persistMetaEntry(statementId, startOffset, endOffset, StatementStatus.PREPARED);
+  }
+
+  @Override
+  public void commit(String statementId) {
+    StatementMeta existing = _statementMetas.get(statementId);
+    if (existing == null) {
+      LOGGER.warn("Attempted to commit unknown statement: {}", statementId);
+      return;
+    }
+    StatementMeta committed = new StatementMeta(
+        statementId, existing._startOffset, existing._endOffset, StatementStatus.COMMITTED);
+    _statementMetas.put(statementId, committed);
+    persistMetaEntry(statementId, existing._startOffset, existing._endOffset, StatementStatus.COMMITTED);
+  }
+
+  @Override
+  public void abort(String statementId) {
+    StatementMeta existing = _statementMetas.get(statementId);
+    long startOffset = existing != null ? existing._startOffset : -1;
+    long endOffset = existing != null ? existing._endOffset : -1;
+    StatementMeta aborted = new StatementMeta(statementId, startOffset, endOffset, StatementStatus.ABORTED);
+    _statementMetas.put(statementId, aborted);
+    persistMetaEntry(statementId, startOffset, endOffset, StatementStatus.ABORTED);
+  }
+
+  @Override
+  public Iterator<byte[]> read(long fromOffset) {
+    return new LogIterator(_logFile, fromOffset, _writeOffset.get());
+  }
+
+  /**
+   * Returns the current write offset (end of log).
+   */
+  public long getWriteOffset() {
+    return _writeOffset.get();
+  }
+
+  /**
+   * Returns the metadata for the given statement, or {@code null} if not found.
+   */
+  public StatementMeta getStatementMeta(String statementId) {
+    return _statementMetas.get(statementId);
+  }
+
+  private void persistMetaEntry(String statementId, long startOffset, long endOffset, StatementStatus status) {
+    try (FileOutputStream fos = new FileOutputStream(_metaFile, true);
+         FileChannel channel = fos.getChannel()) {
+      byte[] line = (statementId + "," + startOffset + "," + endOffset + "," + status.name() + "\n")
+          .getBytes(java.nio.charset.StandardCharsets.UTF_8);
+      channel.write(ByteBuffer.wrap(line));
+      channel.force(true);
+    } catch (IOException e) {
+      String message = "Failed to persist metadata for statement: " + statementId;
+      LOGGER.error(message, e);
+      throw new IllegalStateException(message, e);
+    }
+  }
+
+  private void recoverMetadata() {
+    if (!_metaFile.exists()) {
+      return;
+    }
+    try (BufferedReader reader = new BufferedReader(new FileReader(_metaFile))) {
+      String line;
+      while ((line = reader.readLine()) != null) {

Review Comment:
   `LocalShardLog` persists metadata lines as UTF-8 bytes, but `recoverMetadata()` reads via `FileReader` (platform default charset). This can break recovery on non-UTF-8 default platforms. Read the meta file using an explicit UTF-8 `InputStreamReader` to match the write path.



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/InsertRowApplier.java:
##########
@@ -0,0 +1,191 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.ingest.PreparedStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Applies committed insert data to Pinot's mutable segment path.
+ *
+ * <p>On commit, reads prepared batches from the {@link PreparedStore} and provides them for
+ * indexing into mutable segments. Tracks applied statement IDs in a persistent file to enable
+ * idempotent replay after server restart.
+ *
+ * <p>The applied-statement tracking file is stored at
+ * {@code <dataDir>/insert/applied_statements.log}.
+ *
+ * <p>This class is thread-safe for concurrent apply operations on different statements. Concurrent
+ * apply on the same statement is not expected.
+ */
+public class InsertRowApplier {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(InsertRowApplier.class);
+  private static final String APPLIED_STATEMENTS_FILE = "applied_statements.log";
+
+  private final PreparedStore _preparedStore;
+  private final File _appliedStatementsFile;
+  private final Set<String> _appliedStatements;
+  private final Object _fileLock = new Object();
+
+  /**
+   * Creates a new insert row applier.
+   *
+   * @param preparedStore the prepared store to read batches from
+   * @param dataDir the data directory for storing the applied-statements tracking file
+   */
+  public InsertRowApplier(PreparedStore preparedStore, File dataDir) {
+    _preparedStore = preparedStore;
+    File insertDir = new File(dataDir, "insert");
+    if (!insertDir.exists() && !insertDir.mkdirs()) {
+      throw new RuntimeException("Failed to create insert directory: " + insertDir);
+    }
+    _appliedStatementsFile = new File(insertDir, APPLIED_STATEMENTS_FILE);
+    _appliedStatements = loadAppliedStatements();
+  }
+
+  /**
+   * Applies the committed data for the given statement and partition. Returns the deserialized
+   * rows that should be indexed into the mutable segment.
+   *
+   * <p>This method is idempotent: if the statement has already been applied (tracked in the
+   * applied-statements file), it returns {@code null} to indicate no action is needed.
+   *
+   * @param statementId the statement identifier
+   * @param partitionId the partition identifier
+   * @param sequenceNo the sequence number
+   * @return the list of rows to index, or {@code null} if already applied
+   * @throws IOException if reading or deserializing the prepared data fails
+   */
+  public List<GenericRow> apply(String statementId, int partitionId, long sequenceNo)
+      throws IOException {
+    String applyKey = statementId + ":" + partitionId + ":" + sequenceNo;
+
+    if (_appliedStatements.contains(applyKey)) {
+      LOGGER.info("Statement {} partition {} seq {} already applied, skipping", statementId, partitionId, sequenceNo);
+      return null;
+    }
+
+    byte[] data = _preparedStore.load(statementId, partitionId, sequenceNo);
+    if (data == null) {
+      LOGGER.warn("No prepared data found for statement {} partition {} seq {}", statementId, partitionId, sequenceNo);
+      return null;
+    }
+
+    List<GenericRow> rows = GenericRowSerializer.deserializeRows(data);
+    LOGGER.info("Applied {} rows for statement {} partition {} seq {}", rows.size(), statementId, partitionId,
+        sequenceNo);
+
+    // Record that this batch has been applied
+    markApplied(applyKey);
+
+    return rows;
+  }
+
+  /**
+   * Returns whether a given statement/partition/sequence has already been applied.
+   *
+   * @param statementId the statement identifier
+   * @param partitionId the partition identifier
+   * @param sequenceNo the sequence number
+   * @return {@code true} if already applied
+   */
+  public boolean isApplied(String statementId, int partitionId, long sequenceNo) {
+    String applyKey = statementId + ":" + partitionId + ":" + sequenceNo;
+    return _appliedStatements.contains(applyKey);
+  }
+
+  private void markApplied(String applyKey) {
+    synchronized (_fileLock) {
+      _appliedStatements.add(applyKey);
+      try (BufferedWriter writer = new BufferedWriter(new FileWriter(_appliedStatementsFile, true))) {
+        writer.write(applyKey);
+        writer.newLine();
+        writer.flush();
+      } catch (IOException e) {
+        LOGGER.error("Failed to record applied statement: {}", applyKey, e);
+      }
+    }
+  }
+
+  /**
+   * Removes tracking entries for the given statement from the in-memory set and rewrites the
+   * applied-statements file. Called during GC to prevent the tracking file from growing without
+   * bound.
+   *
+   * @param statementId the statement identifier to remove tracking entries for
+   */
+  public void cleanAppliedEntries(String statementId) {
+    String prefix = statementId + ":";
+    synchronized (_fileLock) {
+      boolean removed = _appliedStatements.removeIf(key -> key.startsWith(prefix));
+      if (removed) {
+        rewriteAppliedStatementsFile();
+        LOGGER.info("Cleaned applied tracking entries for statement: {}", statementId);
+      }
+    }
+  }
+
+  /**
+   * Rewrites the applied-statements file from the current in-memory set.
+   * Must be called under {@code _fileLock}.
+   */
+  private void rewriteAppliedStatementsFile() {
+    try (BufferedWriter writer = new BufferedWriter(new FileWriter(_appliedStatementsFile, false))) {
+      for (String key : _appliedStatements) {
+        writer.write(key);
+        writer.newLine();
+      }
+      writer.flush();
+    } catch (IOException e) {
+      LOGGER.error("Failed to rewrite applied statements file: {}", _appliedStatementsFile, e);
+    }
+  }
+
+  private Set<String> loadAppliedStatements() {
+    Set<String> applied = new HashSet<>();
+    if (!_appliedStatementsFile.exists()) {
+      return applied;
+    }
+    try (BufferedReader reader = new BufferedReader(new FileReader(_appliedStatementsFile))) {
+      String line;

Review Comment:
   `InsertRowApplier` reads the applied-statements log with `FileReader`, which uses the platform default charset. Because the file is persisted and later replayed, it should be read with the same explicit charset used for writing (e.g., UTF-8).



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/InsertRowApplier.java:
##########
@@ -0,0 +1,191 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.ingest.PreparedStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Applies committed insert data to Pinot's mutable segment path.
+ *
+ * <p>On commit, reads prepared batches from the {@link PreparedStore} and provides them for
+ * indexing into mutable segments. Tracks applied statement IDs in a persistent file to enable
+ * idempotent replay after server restart.
+ *
+ * <p>The applied-statement tracking file is stored at
+ * {@code <dataDir>/insert/applied_statements.log}.
+ *
+ * <p>This class is thread-safe for concurrent apply operations on different statements. Concurrent
+ * apply on the same statement is not expected.
+ */
+public class InsertRowApplier {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(InsertRowApplier.class);
+  private static final String APPLIED_STATEMENTS_FILE = "applied_statements.log";
+
+  private final PreparedStore _preparedStore;
+  private final File _appliedStatementsFile;
+  private final Set<String> _appliedStatements;
+  private final Object _fileLock = new Object();
+
+  /**
+   * Creates a new insert row applier.
+   *
+   * @param preparedStore the prepared store to read batches from
+   * @param dataDir the data directory for storing the applied-statements tracking file
+   */
+  public InsertRowApplier(PreparedStore preparedStore, File dataDir) {
+    _preparedStore = preparedStore;
+    File insertDir = new File(dataDir, "insert");
+    if (!insertDir.exists() && !insertDir.mkdirs()) {
+      throw new RuntimeException("Failed to create insert directory: " + insertDir);
+    }
+    _appliedStatementsFile = new File(insertDir, APPLIED_STATEMENTS_FILE);
+    _appliedStatements = loadAppliedStatements();
+  }
+
+  /**
+   * Applies the committed data for the given statement and partition. Returns the deserialized
+   * rows that should be indexed into the mutable segment.
+   *
+   * <p>This method is idempotent: if the statement has already been applied (tracked in the
+   * applied-statements file), it returns {@code null} to indicate no action is needed.
+   *
+   * @param statementId the statement identifier
+   * @param partitionId the partition identifier
+   * @param sequenceNo the sequence number
+   * @return the list of rows to index, or {@code null} if already applied
+   * @throws IOException if reading or deserializing the prepared data fails
+   */
+  public List<GenericRow> apply(String statementId, int partitionId, long sequenceNo)
+      throws IOException {
+    String applyKey = statementId + ":" + partitionId + ":" + sequenceNo;
+
+    if (_appliedStatements.contains(applyKey)) {
+      LOGGER.info("Statement {} partition {} seq {} already applied, skipping", statementId, partitionId, sequenceNo);
+      return null;
+    }
+
+    byte[] data = _preparedStore.load(statementId, partitionId, sequenceNo);
+    if (data == null) {
+      LOGGER.warn("No prepared data found for statement {} partition {} seq {}", statementId, partitionId, sequenceNo);
+      return null;
+    }
+
+    List<GenericRow> rows = GenericRowSerializer.deserializeRows(data);
+    LOGGER.info("Applied {} rows for statement {} partition {} seq {}", rows.size(), statementId, partitionId,
+        sequenceNo);
+
+    // Record that this batch has been applied
+    markApplied(applyKey);
+
+    return rows;
+  }
+
+  /**
+   * Returns whether a given statement/partition/sequence has already been applied.
+   *
+   * @param statementId the statement identifier
+   * @param partitionId the partition identifier
+   * @param sequenceNo the sequence number
+   * @return {@code true} if already applied
+   */
+  public boolean isApplied(String statementId, int partitionId, long sequenceNo) {
+    String applyKey = statementId + ":" + partitionId + ":" + sequenceNo;
+    return _appliedStatements.contains(applyKey);
+  }
+
+  private void markApplied(String applyKey) {
+    synchronized (_fileLock) {
+      _appliedStatements.add(applyKey);
+      try (BufferedWriter writer = new BufferedWriter(new FileWriter(_appliedStatementsFile, true))) {
+        writer.write(applyKey);
+        writer.newLine();
+        writer.flush();
+      } catch (IOException e) {

Review Comment:
   `InsertRowApplier` writes the applied-statements log with `FileWriter`, which uses the platform default charset. Since this file is used across restarts (and potentially across hosts), it should use an explicit charset (e.g., UTF-8) for deterministic behavior.



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/RowInsertExecutor.java:
##########
@@ -0,0 +1,206 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import org.apache.pinot.spi.config.table.TableConfig;
+import org.apache.pinot.spi.data.readers.GenericRow;
+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.PreparedStore;
+import org.apache.pinot.spi.ingest.ShardLog;
+import org.apache.pinot.spi.ingest.ShardLogProvider;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Server-side INSERT INTO executor using a local durable log and prepared store.
+ *
+ * <p><strong>Note (v1 limitation):</strong> This executor requires table configs to be registered
+ * via {@link RowInsertExecutorFactory#registerTableConfig}. In v1, the primary INSERT path uses
+ * {@code ControllerRowInsertExecutor} which resolves table configs from ZooKeeper. This server-side
+ * executor is intended for a future direct-server-ingestion path where the server's
+ * {@code InstanceDataManager} will register table configs as tables are loaded.
+ *
+ * <p>The execution flow for each request:
+ * <ol>
+ *   <li>Validate the request (table exists, rows present, etc.)</li>
+ *   <li>Route rows to target partition(s) based on partition function</li>
+ *   <li>Write rows to the {@link ShardLog} for each partition</li>
+ *   <li>Write prepared batch to the {@link PreparedStore}</li>
+ *   <li>Return {@link InsertResult} with state={@link InsertStatementState#PREPARED}</li>
+ * </ol>
+ *
+ * <p>This class is thread-safe; a single instance may be invoked concurrently from multiple
+ * request threads.
+ */
+public class RowInsertExecutor implements InsertExecutor {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(RowInsertExecutor.class);
+
+  private final ShardLogProvider _shardLogProvider;
+  private final PreparedStore _preparedStore;
+  private final ConcurrentHashMap<String, InsertStatementState> _statementStates = new ConcurrentHashMap<>();
+  private final ConcurrentHashMap<String, TableConfig> _tableConfigs;
+
+  /**
+   * Creates a new row insert executor.
+   *
+   * @param shardLogProvider the provider for shard logs
+   * @param preparedStore the prepared store for staging data
+   * @param tableConfigs a map of table name to table config for validation and partition routing
+   */
+  public RowInsertExecutor(ShardLogProvider shardLogProvider, PreparedStore preparedStore,
+      ConcurrentHashMap<String, TableConfig> tableConfigs) {
+    _shardLogProvider = shardLogProvider;
+    _preparedStore = preparedStore;
+    _tableConfigs = tableConfigs;
+  }
+
+  @Override
+  public InsertResult execute(InsertRequest request) {
+    String statementId = request.getStatementId();
+    String tableName = request.getTableName();
+    List<GenericRow> rows = request.getRows();
+
+    LOGGER.info("Executing INSERT for statement {} on table {} with {} rows", statementId, tableName, rows.size());
+
+    // Validate the request
+    if (tableName == null || tableName.isEmpty()) {
+      return buildErrorResult(statementId, "Table name is required", "INVALID_TABLE");
+    }
+    if (rows == null || rows.isEmpty()) {
+      return buildErrorResult(statementId, "At least one row is required for ROW insert", "EMPTY_ROWS");
+    }
+
+    TableConfig tableConfig = _tableConfigs.get(tableName);
+    if (tableConfig == null) {
+      return buildErrorResult(statementId, "Table not found: " + tableName, "TABLE_NOT_FOUND");
+    }
+
+    // Create partition router
+    InsertPartitionRouter router;
+    try {
+      router = new InsertPartitionRouter(tableConfig);
+    } catch (IllegalArgumentException e) {
+      return buildErrorResult(statementId, e.getMessage(), "PARTITION_CONFIG_ERROR");
+    }
+
+    // Route rows to partitions
+    Map<Integer, List<GenericRow>> partitionedRows = router.routeRows(rows);
+
+    _statementStates.put(statementId, InsertStatementState.ACCEPTED);
+
+    // Write rows to shard log and prepared store for each partition
+    try {
+      for (Map.Entry<Integer, List<GenericRow>> entry : partitionedRows.entrySet()) {
+        int partitionId = entry.getKey();
+        List<GenericRow> partitionRows = entry.getValue();
+
+        ShardLog shardLog = _shardLogProvider.getShardLog(tableName, partitionId);
+
+        // Serialize and append rows to the shard log
+        byte[] serializedRows = GenericRowSerializer.serializeRows(partitionRows);
+        long startOffset = shardLog.append(serializedRows);
+        long endOffset = startOffset;  // Single append, so start == end
+
+        // Also write to prepared store for durability
+        _preparedStore.store(statementId, partitionId, 0, serializedRows);
+
+        // Mark the offset range as prepared in the shard log
+        shardLog.prepare(statementId, startOffset, endOffset);
+      }
+
+      _statementStates.put(statementId, InsertStatementState.PREPARED);
+
+      LOGGER.info("INSERT statement {} prepared with {} partition(s)", statementId, partitionedRows.size());
+
+      return new InsertResult.Builder()
+          .setStatementId(statementId)
+          .setState(InsertStatementState.PREPARED)
+          .setMessage("Insert prepared with " + rows.size() + " rows across "
+              + partitionedRows.size() + " partition(s)")
+          .build();
+    } catch (Exception e) {
+      LOGGER.error("Failed to execute INSERT for statement {}", statementId, e);
+      _statementStates.put(statementId, InsertStatementState.ABORTED);
+      return buildErrorResult(statementId, "Failed to write rows: " + e.getMessage(), "WRITE_ERROR");
+    }
+  }
+
+  @Override
+  public InsertResult getStatus(String statementId) {
+    InsertStatementState state = _statementStates.get(statementId);
+    if (state == null) {
+      return new InsertResult.Builder()
+          .setStatementId(statementId)
+          .setState(InsertStatementState.ABORTED)
+          .setMessage("Unknown statement: " + statementId)
+          .build();
+    }
+    return new InsertResult.Builder()
+        .setStatementId(statementId)
+        .setState(state)
+        .build();
+  }
+
+  @Override
+  public InsertResult abort(String statementId) {
+    InsertStatementState currentState = _statementStates.get(statementId);
+    if (currentState == null) {
+      return new InsertResult.Builder()
+          .setStatementId(statementId)
+          .setState(InsertStatementState.ABORTED)
+          .setMessage("Unknown statement: " + statementId)
+          .build();
+    }
+
+    _statementStates.put(statementId, InsertStatementState.ABORTED);
+    _preparedStore.cleanup(statementId);
+

Review Comment:
   `abort()` only updates the in-memory state and cleans the prepared store, but it never calls `ShardLog.abort(statementId)` for the affected partitions. That leaves shard-log metadata in PREPARED/COMMITTED and can confuse recovery or future GC. Consider tracking the partitions touched by each statement (during `execute`) so `abort()` can abort/rollback those shard logs as well.



##########
pinot-core/src/main/java/org/apache/pinot/core/data/manager/ingest/InsertRowApplier.java:
##########
@@ -0,0 +1,191 @@
+/**
+ * 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.core.data.manager.ingest;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileReader;
+import java.io.FileWriter;
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import org.apache.pinot.spi.data.readers.GenericRow;
+import org.apache.pinot.spi.ingest.PreparedStore;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+
+/**
+ * Applies committed insert data to Pinot's mutable segment path.
+ *
+ * <p>On commit, reads prepared batches from the {@link PreparedStore} and provides them for
+ * indexing into mutable segments. Tracks applied statement IDs in a persistent file to enable
+ * idempotent replay after server restart.
+ *
+ * <p>The applied-statement tracking file is stored at
+ * {@code <dataDir>/insert/applied_statements.log}.
+ *
+ * <p>This class is thread-safe for concurrent apply operations on different statements. Concurrent
+ * apply on the same statement is not expected.
+ */
+public class InsertRowApplier {
+
+  private static final Logger LOGGER = LoggerFactory.getLogger(InsertRowApplier.class);
+  private static final String APPLIED_STATEMENTS_FILE = "applied_statements.log";
+
+  private final PreparedStore _preparedStore;
+  private final File _appliedStatementsFile;
+  private final Set<String> _appliedStatements;
+  private final Object _fileLock = new Object();
+

Review Comment:
   `InsertRowApplier` claims thread-safety for concurrent applies, but `_appliedStatements` is a plain `HashSet` that is read outside the `_fileLock` (e.g., `contains()` in `apply()`/`isApplied()`) while being mutated under the lock. This can lead to data races and `HashSet` corruption. Use a concurrent set (e.g., `ConcurrentHashMap.newKeySet()`) or synchronize all accesses to `_appliedStatements` (reads and writes).



-- 
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: commits-unsubscribe@pinot.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org