li3zhi4 commented on code in PR #11746:
URL: https://github.com/apache/seatunnel/pull/11746#discussion_r3766022591
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java:
##########
@@ -106,6 +115,324 @@ private Collection<JdbcSourceSplit> createDynamicSplits(
return splits;
}
+ private static final String COMPOSITE_KEY_SEPARATOR = ",";
+
+ /**
+ * Split a table on the full composite key tuple. Boundaries are
tuple-ordered ((start, end] per
+ * chunk, matching {@link #buildCompositeCondition}), so composite-key
tables whose first column
+ * repeats heavily still split into balanced chunks via the remaining key
columns.
+ *
+ * <p><b>Performance trade-off:</b> unlike the single-column paths which
compute chunk
+ * boundaries arithmetically (closed-form evenly/unevenly sized chunks),
this path walks the
+ * boundary with one database round trip per chunk ({@link
#queryNextChunkMaxComposite} scans at
+ * most {@code split.size} rows per call). For a large table with many
chunks this is more round
+ * trips than the single-column arithmetic path, in exchange for correct,
balanced chunks on
+ * composite keys where arithmetic is impossible (the first column has low
cardinality).
+ * Boundary queries are index-friendly (expanded OR/AND), so each round
trip is cheap.
+ */
+ private Collection<JdbcSourceSplit> createCompositeDynamicSplits(
+ JdbcSourceTable table, SeaTunnelRowType splitKey) throws
SQLException {
+ String[] columns = splitKey.getFieldNames();
+ Object[][] minMax = queryMinMaxComposite(table, columns);
+ Object[] min = minMax[0];
+ Object[] max = minMax[1];
+ if (min == null || max == null) {
+ // empty table, return a single full-table split
+ return Collections.singletonList(createCompositeSplit(table, 0,
splitKey, null, null));
+ }
+
+ // first chunk end
+ Object[] firstEnd = queryNextChunkMaxComposite(table, columns,
config.getSplitSize(), min);
+ if (firstEnd == null || compareArrays(firstEnd, max) >= 0) {
+ // all data fits in one chunk
+ return Collections.singletonList(createCompositeSplit(table, 0,
splitKey, null, null));
+ }
+
+ List<JdbcSourceSplit> splits = new ArrayList<>();
+ Object[] chunkStart = null;
+ Object[] chunkEnd = firstEnd;
+ int index = 0;
+ while (chunkEnd != null && compareArrays(chunkEnd, max) < 0) {
+ splits.add(createCompositeSplit(table, index++, splitKey,
chunkStart, chunkEnd));
+ chunkStart = chunkEnd;
+ chunkEnd =
+ queryNextChunkMaxComposite(table, columns,
config.getSplitSize(), chunkStart);
+ if (chunkEnd != null && Arrays.equals(chunkStart, chunkEnd)) {
+ // we don't allow equal chunk start and end,
+ // should query the next one larger than chunkEnd
+ chunkEnd = queryMinComposite(table, columns, chunkEnd);
+ }
+ }
+ // add the ending split
+ splits.add(createCompositeSplit(table, index, splitKey, chunkStart,
null));
+ return splits;
+ }
+
+ private JdbcSourceSplit createCompositeSplit(
+ JdbcSourceTable table,
+ int index,
+ SeaTunnelRowType splitKey,
+ Object[] start,
+ Object[] end) {
+ String joinedNames = String.join(COMPOSITE_KEY_SEPARATOR,
splitKey.getFieldNames());
+ return new JdbcSourceSplit(
+ table.getTablePath(),
+ createSplitId(table.getTablePath(), index),
+ table.getQuery(),
+ joinedNames,
+ splitKey,
+ start,
+ end);
+ }
+
+ /**
+ * Queries the tuple-ordered minimum and maximum of the composite key
using {@code ORDER BY ...
+ * ASC/DESC LIMIT 1} (index-friendly, works for any comparable type and
avoids MIN()/MAX()
+ * aggregates which would scan without index benefit).
+ *
+ * @return {@code [minRow, maxRow]}, each element null when the table is
empty
+ */
+ private Object[][] queryMinMaxComposite(JdbcSourceTable table, String[]
columns)
+ throws SQLException {
+ StringBuilder selectCols = new StringBuilder();
+ StringBuilder orderAsc = new StringBuilder();
+ StringBuilder orderDesc = new StringBuilder();
+ for (int i = 0; i < columns.length; i++) {
+ if (i > 0) {
+ selectCols.append(", ");
+ orderAsc.append(", ");
+ orderDesc.append(", ");
+ }
+ String quoted = jdbcDialect.quoteIdentifier(columns[i]);
+ selectCols.append(quoted);
+ orderAsc.append(quoted).append(" ASC");
+ orderDesc.append(quoted).append(" DESC");
+ }
+
+ String fromClause;
+ if (StringUtils.isNotBlank(table.getQuery())) {
+ fromClause = "(" + normalizeQuery(table.getQuery()) + ") AS
_st_tmp";
+ } else {
+ fromClause = jdbcDialect.tableIdentifier(table.getTablePath());
+ }
+
+ String minQuery =
+ "SELECT "
+ + selectCols
+ + " FROM "
+ + fromClause
+ + " ORDER BY "
+ + orderAsc
+ + jdbcDialect.getLimitClause(1);
+ String maxQuery =
+ "SELECT "
+ + selectCols
+ + " FROM "
+ + fromClause
+ + " ORDER BY "
+ + orderDesc
+ + jdbcDialect.getLimitClause(1);
+
+ Connection conn = getOrEstablishConnection();
+ Object[] minRow = executeRowQuery(conn, minQuery, columns.length);
+ Object[] maxRow = executeRowQuery(conn, maxQuery, columns.length);
+ return new Object[][] {minRow, maxRow};
+ }
+
+ /**
+ * Walks the composite key boundary: returns the last row of the first
{@code chunkSize} rows
+ * strictly greater than {@code includedLowerBound} (tuple-ordered), using
an expanded OR/AND
+ * condition so the composite index is used instead of a row-constructor
comparison. Each call
+ * scans at most {@code chunkSize} rows.
+ *
+ * @return the next chunk-end tuple, or null when no row is greater than
the bound
+ */
+ private Object[] queryNextChunkMaxComposite(
+ JdbcSourceTable table, String[] columns, int chunkSize, Object[]
includedLowerBound)
+ throws SQLException {
+ StringBuilder columnList = new StringBuilder();
+ StringBuilder orderBy = new StringBuilder();
+ for (int i = 0; i < columns.length; i++) {
+ if (i > 0) {
+ columnList.append(", ");
+ orderBy.append(", ");
+ }
+ String quoted = jdbcDialect.quoteIdentifier(columns[i]);
+ columnList.append(quoted);
+ orderBy.append(quoted);
+ }
+
+ // Expanded OR condition for index usage:
+ // (a > ?) OR (a = ? AND b > ?) OR (a = ? AND b = ? AND c >= ?)
+ StringBuilder where = new StringBuilder("(");
+ for (int i = 0; i < columns.length; i++) {
+ if (i > 0) {
+ where.append(" OR ");
+ }
+ where.append("(");
+ for (int j = 0; j <= i; j++) {
+ if (j > 0) {
+ where.append(" AND ");
+ }
+ String quoted = jdbcDialect.quoteIdentifier(columns[j]);
+ if (j < i) {
+ where.append(quoted).append(" = ?");
+ } else {
+ String op = (i == columns.length - 1) ? ">=" : ">";
+ where.append(quoted).append(" ").append(op).append(" ?");
+ }
+ }
+ where.append(")");
+ }
+ where.append(")");
+
+ String fromClause;
+ if (StringUtils.isNotBlank(table.getQuery())) {
+ fromClause = "(" + normalizeQuery(table.getQuery()) + ") AS
_st_tmp";
+ } else {
+ fromClause = jdbcDialect.tableIdentifier(table.getTablePath());
+ }
+
+ String sql =
+ "SELECT "
+ + columnList
+ + " FROM "
+ + fromClause
+ + " WHERE "
+ + where
+ + " ORDER BY "
+ + orderBy
+ + " ASC"
+ + jdbcDialect.getLimitClause(chunkSize);
+
+ Connection conn = getOrEstablishConnection();
+ try (java.sql.PreparedStatement ps = conn.prepareStatement(sql)) {
+ // Bind: v1, v1,v2, v1,v2,v3 ... (cumulative per OR branch)
+ int paramIndex = 1;
+ for (int i = 0; i < columns.length; i++) {
+ for (int j = 0; j <= i; j++) {
+ ps.setObject(paramIndex++, includedLowerBound[j]);
+ }
+ }
+ try (ResultSet rs = ps.executeQuery()) {
+ Object[] lastRow = null;
+ while (rs.next()) {
+ lastRow = new Object[columns.length];
+ for (int i = 0; i < columns.length; i++) {
+ lastRow[i] = rs.getObject(i + 1);
+ }
+ }
+ return lastRow;
+ }
+ }
+ }
+
+ /**
+ * Returns the first tuple strictly greater than {@code
excludedLowerBound} (tuple-ordered).
+ * Used to advance past a duplicated chunk boundary value so chunks never
degenerate into
+ * zero-row ranges.
+ *
+ * @return the next greater tuple, or null when none exists
+ */
+ private Object[] queryMinComposite(
+ JdbcSourceTable table, String[] columns, Object[]
excludedLowerBound)
+ throws SQLException {
+ StringBuilder columnList = new StringBuilder();
+ StringBuilder orderBy = new StringBuilder();
+ for (int i = 0; i < columns.length; i++) {
+ if (i > 0) {
+ columnList.append(", ");
+ orderBy.append(", ");
+ }
+ String quoted = jdbcDialect.quoteIdentifier(columns[i]);
+ columnList.append(quoted);
+ orderBy.append(quoted);
+ }
+
+ String fromClause;
+ if (StringUtils.isNotBlank(table.getQuery())) {
+ fromClause = "(" + normalizeQuery(table.getQuery()) + ") AS
_st_tmp";
Review Comment:
Thanks - addressed in the latest commits. The composite boundary queries now
use "FROM (...) _st_tmp" without AS (inline-view alias per Oracle
table_reference grammar), matching the single-column path which already used
the AS-less form. FETCH FIRST requires Oracle 12c Release 1 - documented on
OracleDialect.getLimitClause and supportCompositeKeySplit, and the dialect only
enables composite splitting after E2E validation. Added JdbcOracleSplitIT
covering composite-PK splitting (300 rows read exactly once) and a custom-query
case (single split by design, all rows readable); getLimitClause also has the
Oracle 12c+ caveat in its Javadoc.
##########
seatunnel-connectors-v2/connector-jdbc/src/main/java/org/apache/seatunnel/connectors/seatunnel/jdbc/source/DynamicChunkSplitter.java:
##########
@@ -106,6 +115,324 @@ private Collection<JdbcSourceSplit> createDynamicSplits(
return splits;
}
+ private static final String COMPOSITE_KEY_SEPARATOR = ",";
+
+ /**
+ * Split a table on the full composite key tuple. Boundaries are
tuple-ordered ((start, end] per
+ * chunk, matching {@link #buildCompositeCondition}), so composite-key
tables whose first column
+ * repeats heavily still split into balanced chunks via the remaining key
columns.
+ *
+ * <p><b>Performance trade-off:</b> unlike the single-column paths which
compute chunk
+ * boundaries arithmetically (closed-form evenly/unevenly sized chunks),
this path walks the
+ * boundary with one database round trip per chunk ({@link
#queryNextChunkMaxComposite} scans at
+ * most {@code split.size} rows per call). For a large table with many
chunks this is more round
+ * trips than the single-column arithmetic path, in exchange for correct,
balanced chunks on
+ * composite keys where arithmetic is impossible (the first column has low
cardinality).
+ * Boundary queries are index-friendly (expanded OR/AND), so each round
trip is cheap.
+ */
+ private Collection<JdbcSourceSplit> createCompositeDynamicSplits(
+ JdbcSourceTable table, SeaTunnelRowType splitKey) throws
SQLException {
+ String[] columns = splitKey.getFieldNames();
+ Object[][] minMax = queryMinMaxComposite(table, columns);
+ Object[] min = minMax[0];
+ Object[] max = minMax[1];
+ if (min == null || max == null) {
+ // empty table, return a single full-table split
+ return Collections.singletonList(createCompositeSplit(table, 0,
splitKey, null, null));
+ }
+
+ // first chunk end
+ Object[] firstEnd = queryNextChunkMaxComposite(table, columns,
config.getSplitSize(), min);
Review Comment:
Thanks for the careful analysis. The boundary walk is indeed O(N) in total,
so let me share what we verified and how we framed it:
1. In the target scenario (composite primary key whose first column repeats
heavily), the previous single-column path is strictly worse: when the first key
column is constant, min==max and the splitter falls back to a single
full-table-scan chunk with no parallelism or boundaries. The tuple-ordered seek
is what enables parallel split reads in this scenario at all; its O(N) is split
into index-friendly range scans of at most split.size rows per call, each
interruptible and streamable, rather than one unbounded full-table scan.
Sampling-based boundary estimation is not an option here - dev itself makes
sampling opt-in/disableable (split.allow-sampling) because it scans the full
column.
2. For composite keys whose first column is high-cardinality, the
single-column arithmetic path remains available via partition_column, so such
tables are not forced into the seek path.
3. Network/temporary-object saving: we added testBoundaryQueryOffsetForm to
the E2E of all five dialects (MySQL, PostgreSQL, SQLite, SQL Server, Oracle)
verifying that the boundary can be expressed as LIMIT 1 OFFSET chunkSize-1 (or
OFFSET ... ROWS FETCH NEXT on SQL Server / Oracle 12c+), returning exactly one
row equal to the last row of LIMIT chunkSize - a candidate follow-up
optimization.
Streaming boundary generation (producing the next boundary while readers
consume the current chunk) is a possible further improvement; it concerns
split-generation timing and could be tracked separately. Happy to iterate
further if you have specific concerns.
--
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]
