LeonYoah commented on code in PR #11746:
URL: https://github.com/apache/seatunnel/pull/11746#discussion_r3763507659
##########
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:
If Oracle is intended to be supported, please fix the generated SQL.
Oracle’s official [`table_reference`
grammar](https://docs.oracle.com/en/database/oracle/oracle-database/12.2/sqlrf/img_text/table_reference.html)
specifies `query_table_expression [t_alias]`, so inline views must use `FROM
(...) alias`, without `AS`. Also, [`FETCH FIRST` was introduced in Oracle 12c
Release 1](https://docs.oracle.com/database/121/NEWFT/chapter12101.htm). Please
add Oracle integration tests covering composite-key splitting, custom queries,
and the supported Oracle versions.
--
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]
