This is an automated email from the ASF dual-hosted git repository.
thomasrebele pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/calcite.git
The following commit(s) were added to refs/heads/main by this push:
new bf070e440d [CALCITE-7650] Bytecode by JDK 11 is invalid due to
Checkerframework annotations
bf070e440d is described below
commit bf070e440d88d551bbf82ea8f48c67857c014e8b
Author: Thomas Rebele <[email protected]>
AuthorDate: Fri Jul 10 17:09:01 2026 +0200
[CALCITE-7650] Bytecode by JDK 11 is invalid due to Checkerframework
annotations
---
.../apache/calcite/adapter/clone/ArrayTable.java | 15 +-
.../calcite/adapter/enumerable/EnumUtils.java | 60 ++++-
.../enumerable/EnumerableInterpretable.java | 62 ++++--
.../apache/calcite/interpreter/Interpreter.java | 18 +-
.../org/apache/calcite/runtime/SqlFunctions.java | 20 +-
.../calcite/schema/impl/ListTransientTable.java | 81 ++++---
.../apache/calcite/sql2rel/SqlToRelConverter.java | 61 +++---
.../main/java/org/apache/calcite/util/Holder.java | 2 +-
.../org/apache/calcite/util/mapping/Mappings.java | 27 ++-
.../calcite/adapter/csv/CsvFilterableTable.java | 28 ++-
.../calcite/adapter/csv/CsvScannableTable.java | 26 ++-
.../adapter/csv/CsvStreamScannableTable.java | 26 ++-
.../org/apache/calcite/example/maze/MazeTable.java | 31 ++-
.../calcite/adapter/file/JsonScannableTable.java | 21 +-
.../geode/simple/GeodeSimpleScannableTable.java | 35 ++-
.../calcite/adapter/kafka/KafkaStreamTable.java | 57 +++--
.../apache/calcite/linq4j/EnumerableDefaults.java | 49 +++--
.../calcite/adapter/os/FilesTableFunction.java | 243 ++++++++++++---------
.../adapter/os/GitCommitsTableFunction.java | 179 ++++++++-------
.../apache/calcite/adapter/tpcds/TpcdsSchema.java | 67 +++---
20 files changed, 680 insertions(+), 428 deletions(-)
diff --git
a/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
b/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
index 8400ac21d6..fabbe6ea8d 100644
--- a/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
+++ b/core/src/main/java/org/apache/calcite/adapter/clone/ArrayTable.java
@@ -87,12 +87,15 @@ class ArrayTable extends AbstractQueryableTable implements
ScannableTable {
}
@Override public Enumerable<@Nullable Object[]> scan(DataContext root) {
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Content content = supplier.get();
- return content.arrayEnumerator();
- }
- };
+ return new ArrayTableEnumerable();
+ }
+
+ /** Enumerable for {@link ArrayTable}. */
+ private class ArrayTableEnumerable extends AbstractEnumerable<@Nullable
Object[]> {
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Content content = supplier.get();
+ return content.arrayEnumerator();
+ }
}
@Override public <T> Queryable<T> asQueryable(final QueryProvider
queryProvider,
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
index 129180e584..80c46ec3f9 100644
--- a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
+++ b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumUtils.java
@@ -1086,12 +1086,29 @@ static Expression tumblingWindowSelector(
public static Enumerable<@Nullable Object[]> sessionize(
Enumerator<@Nullable Object[]> inputEnumerator,
int indexOfWatermarkedColumn, int indexOfKeyColumn, long gap) {
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new SessionizationEnumerator(inputEnumerator,
- indexOfWatermarkedColumn, indexOfKeyColumn, gap);
- }
- };
+ return new SessionizeEnumerable(inputEnumerator, indexOfWatermarkedColumn,
indexOfKeyColumn,
+ gap);
+ }
+
+ /** Enumerable for {@link #sessionize(Enumerator, int, int, long)}. */
+ private static class SessionizeEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final Enumerator<@Nullable Object[]> inputEnumerator;
+ private final int indexOfWatermarkedColumn;
+ private final int indexOfKeyColumn;
+ private final long gap;
+
+ SessionizeEnumerable(Enumerator<@Nullable Object[]> inputEnumerator,
+ int indexOfWatermarkedColumn, int indexOfKeyColumn, long gap) {
+ this.inputEnumerator = inputEnumerator;
+ this.indexOfWatermarkedColumn = indexOfWatermarkedColumn;
+ this.indexOfKeyColumn = indexOfKeyColumn;
+ this.gap = gap;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new SessionizationEnumerator(inputEnumerator,
+ indexOfWatermarkedColumn, indexOfKeyColumn, gap);
+ }
}
/** Enumerator that converts rows into sessions separated by gaps. */
@@ -1234,12 +1251,31 @@ private static Pair<Long, Long> computeInitWindow(long
ts, long gap) {
public static Enumerable<@Nullable Object[]> hopping(
Enumerator<@Nullable Object[]> inputEnumerator,
int indexOfWatermarkedColumn, long emitFrequency, long windowSize, long
offset) {
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new HopEnumerator(inputEnumerator,
- indexOfWatermarkedColumn, emitFrequency, windowSize, offset);
- }
- };
+ return new HoppingEnumerable(inputEnumerator, indexOfWatermarkedColumn,
emitFrequency,
+ windowSize, offset);
+ }
+
+ /** Enumerable for {@link #hopping(Enumerator, int, long, long, long)}. */
+ private static class HoppingEnumerable extends AbstractEnumerable<@Nullable
Object[]> {
+ private final Enumerator<@Nullable Object[]> inputEnumerator;
+ private final int indexOfWatermarkedColumn;
+ private final long emitFrequency;
+ private final long windowSize;
+ private final long offset;
+
+ HoppingEnumerable(Enumerator<@Nullable Object[]> inputEnumerator,
+ int indexOfWatermarkedColumn, long emitFrequency, long windowSize,
long offset) {
+ this.inputEnumerator = inputEnumerator;
+ this.indexOfWatermarkedColumn = indexOfWatermarkedColumn;
+ this.emitFrequency = emitFrequency;
+ this.windowSize = windowSize;
+ this.offset = offset;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new HopEnumerator(inputEnumerator,
+ indexOfWatermarkedColumn, emitFrequency, windowSize, offset);
+ }
}
/** Enumerator that computes HOP. */
diff --git
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
index 5f32ab1c71..7ed858b754 100644
---
a/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
+++
b/core/src/main/java/org/apache/calcite/adapter/enumerable/EnumerableInterpretable.java
@@ -205,32 +205,50 @@ static ArrayBindable box(final Bindable bindable) {
@Override public Enumerable<@Nullable Object[]> bind(DataContext
dataContext) {
final Enumerable<?> enumerable = bindable.bind(dataContext);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Enumerator<?> enumerator = enumerable.enumerator();
- return new Enumerator<@Nullable Object[]>() {
- @Override public @Nullable Object[] current() {
- return new Object[] {enumerator.current()};
- }
-
- @Override public boolean moveNext() {
- return enumerator.moveNext();
- }
-
- @Override public void reset() {
- enumerator.reset();
- }
-
- @Override public void close() {
- enumerator.close();
- }
- };
- }
- };
+ return new BoxEnumerable(enumerable);
}
};
}
+ /** Enumerable for {@link #box(Bindable)}. */
+ private static class BoxEnumerable extends AbstractEnumerable<@Nullable
Object[]> {
+ private final Enumerable<?> enumerable;
+
+ BoxEnumerable(Enumerable<?> enumerable) {
+ this.enumerable = enumerable;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Enumerator<?> enumerator = enumerable.enumerator();
+ return new BoxEnumerator(enumerator);
+ }
+ }
+
+ /** Enumerator for {@link #box(Bindable)}. */
+ private static class BoxEnumerator implements Enumerator<@Nullable Object[]>
{
+ private final Enumerator<?> enumerator;
+
+ BoxEnumerator(Enumerator<?> enumerator) {
+ this.enumerator = enumerator;
+ }
+
+ @Override public @Nullable Object[] current() {
+ return new Object[] { enumerator.current()};
+ }
+
+ @Override public boolean moveNext() {
+ return enumerator.moveNext();
+ }
+
+ @Override public void reset() {
+ enumerator.reset();
+ }
+
+ @Override public void close() {
+ enumerator.close();
+ }
+ }
+
/** Interpreter node that reads from an {@link Enumerable}.
*
* <p>From the interpreter's perspective, it is a leaf node. */
diff --git a/core/src/main/java/org/apache/calcite/interpreter/Interpreter.java
b/core/src/main/java/org/apache/calcite/interpreter/Interpreter.java
index 6bd8bde4c1..8bc82646a7 100644
--- a/core/src/main/java/org/apache/calcite/interpreter/Interpreter.java
+++ b/core/src/main/java/org/apache/calcite/interpreter/Interpreter.java
@@ -118,11 +118,19 @@ private static RelNode optimize(RelNode rootRel) {
rows = Linq4j.iterableEnumerator(queue);
}
- return new TransformedEnumerator<Row, @Nullable Object[]>(rows) {
- @Override protected @Nullable Object[] transform(Row row) {
- return row.getValues();
- }
- };
+ return new InterpreterEnumerator(rows);
+ }
+
+ /** Enumerator for {@link Interpreter}. */
+ private static class InterpreterEnumerator
+ extends TransformedEnumerator<Row, @Nullable Object[]> {
+ InterpreterEnumerator(Enumerator<Row> rows) {
+ super(rows);
+ }
+
+ @Override protected @Nullable Object[] transform(Row row) {
+ return row.getValues();
+ }
}
@SuppressWarnings("CatchAndPrintStackTrace")
diff --git a/core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java
b/core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java
index 5dfad1da34..ed6a3ba16f 100644
--- a/core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java
+++ b/core/src/main/java/org/apache/calcite/runtime/SqlFunctions.java
@@ -222,11 +222,21 @@ public class SqlFunctions {
enumerators.add(Linq4j.enumerator((List) list));
}
final Enumerator<List<@Nullable Object>> product =
Linq4j.product(enumerators);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return Linq4j.transform(product, List::toArray);
- }
- };
+ return new ArrayCartesianProductEnumerable(product);
+ }
+
+ /** Enumerable for {@link #arrayCartesianProduct(Object[])}. */
+ private static class ArrayCartesianProductEnumerable
+ extends AbstractEnumerable<@Nullable Object[]> {
+ private final Enumerator<List<@Nullable Object>> product;
+
+ ArrayCartesianProductEnumerable(Enumerator<List<@Nullable Object>>
product) {
+ this.product = product;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return Linq4j.transform(product, List::toArray);
+ }
}
/** Holds, for each thread, a map from sequence name to sequence current
diff --git
a/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
b/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
index a80556ceef..f04c9372b7 100644
--- a/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
+++ b/core/src/main/java/org/apache/calcite/schema/impl/ListTransientTable.java
@@ -93,38 +93,57 @@ public ListTransientTable(String name, RelDataType rowType)
{
final AtomicBoolean cancelFlag =
DataContext.Variable.CANCEL_FLAG.get(root);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new Enumerator<@Nullable Object[]>() {
- @SuppressWarnings({"rawtypes", "unchecked"})
- private final List list = new ArrayList(rows);
- private int i = -1;
-
- // TODO cleaner way to handle non-array objects?
- @Override public Object[] current() {
- Object current = list.get(i);
- return current != null && current.getClass().isArray()
- ? (Object[]) current
- : new Object[]{current};
- }
-
- @Override public boolean moveNext() {
- if (cancelFlag != null && cancelFlag.get()) {
- return false;
- }
-
- return ++i < list.size();
- }
-
- @Override public void reset() {
- i = -1;
- }
-
- @Override public void close() {
- }
- };
+ return new ListTransientTableEnumerable(cancelFlag);
+ }
+
+ /** Enumerable for {@link ListTransientTable}. */
+ private class ListTransientTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final @Nullable AtomicBoolean cancelFlag;
+
+ ListTransientTableEnumerable(@Nullable AtomicBoolean cancelFlag) {
+ this.cancelFlag = cancelFlag;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new ListTransientTableEnumerator(cancelFlag);
+ }
+ }
+
+ /** Enumerator for {@link ListTransientTable}. */
+ private class ListTransientTableEnumerator implements Enumerator<@Nullable
Object[]> {
+ @SuppressWarnings({"rawtypes", "unchecked"})
+ private final List list;
+ private final @Nullable AtomicBoolean cancelFlag;
+ private int i;
+
+ ListTransientTableEnumerator(@Nullable AtomicBoolean cancelFlag) {
+ this.cancelFlag = cancelFlag;
+ list = new ArrayList(rows);
+ i = -1;
+ }
+
+ // TODO cleaner way to handle non-array objects?
+ @Override public Object[] current() {
+ Object current = list.get(i);
+ return current != null && current.getClass().isArray()
+ ? (Object[]) current
+ : new Object[]{current};
+ }
+
+ @Override public boolean moveNext() {
+ if (cancelFlag != null && cancelFlag.get()) {
+ return false;
}
- };
+
+ return ++i < list.size();
+ }
+
+ @Override public void reset() {
+ i = -1;
+ }
+
+ @Override public void close() {
+ }
}
@Override public Expression getExpression(SchemaPlus schema, String
tableName,
diff --git
a/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
b/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
index f8b661c53f..182ed7a254 100644
--- a/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
+++ b/core/src/main/java/org/apache/calcite/sql2rel/SqlToRelConverter.java
@@ -2967,32 +2967,7 @@ protected void convertMatchRecognize(Blackboard bb,
final Set<String> patternVarsSet = new HashSet<>();
SqlNode pattern = matchRecognize.getPattern();
final SqlBasicVisitor<@Nullable RexNode> patternVarVisitor =
- new SqlBasicVisitor<@Nullable RexNode>() {
- @Override public RexNode visit(SqlCall call) {
- List<SqlNode> operands = call.getOperandList();
- List<RexNode> newOperands = new ArrayList<>();
- for (SqlNode node : operands) {
- RexNode arg = requireNonNull(node.accept(this), node::toString);
- newOperands.add(arg);
- }
- return rexBuilder.makeCall(call.getParserPosition(),
- validator().getUnknownType(), call.getOperator(), newOperands);
- }
-
- @Override public RexNode visit(SqlIdentifier id) {
- assert id.isSimple();
- patternVarsSet.add(id.getSimple());
- return rexBuilder.makeLiteral(id.getSimple());
- }
-
- @Override public RexNode visit(SqlLiteral literal) {
- if (literal instanceof SqlNumericLiteral) {
- return
rexBuilder.makeExactLiteral(BigDecimal.valueOf(literal.intValue(true)));
- } else {
- return rexBuilder.makeLiteral(literal.booleanValue());
- }
- }
- };
+ new PatternVarVisitor(patternVarsSet);
final RexNode patternNode = pattern.accept(patternVarVisitor);
if (patternNode == null) {
throw new AssertionError("pattern is not found in " + pattern);
@@ -7124,4 +7099,38 @@ private class MeasureBlackboard extends Blackboard {
return super.lookupMeasure(identifier);
}
}
+
+ /** Visitor for {@link #convertMatchRecognize(Blackboard,
SqlMatchRecognize)}. */
+ private class PatternVarVisitor extends SqlBasicVisitor<@Nullable RexNode> {
+ private final Set<String> patternVarsSet;
+
+ PatternVarVisitor(Set<String> patternVarsSet) {
+ this.patternVarsSet = patternVarsSet;
+ }
+
+ @Override public RexNode visit(SqlCall call) {
+ List<SqlNode> operands = call.getOperandList();
+ List<RexNode> newOperands = new ArrayList<>();
+ for (SqlNode node : operands) {
+ RexNode arg = requireNonNull(node.accept(this), node::toString);
+ newOperands.add(arg);
+ }
+ return rexBuilder.makeCall(call.getParserPosition(),
+ validator().getUnknownType(), call.getOperator(), newOperands);
+ }
+
+ @Override public RexNode visit(SqlIdentifier id) {
+ assert id.isSimple();
+ patternVarsSet.add(id.getSimple());
+ return rexBuilder.makeLiteral(id.getSimple());
+ }
+
+ @Override public RexNode visit(SqlLiteral literal) {
+ if (literal instanceof SqlNumericLiteral) {
+ return
rexBuilder.makeExactLiteral(BigDecimal.valueOf(literal.intValue(true)));
+ } else {
+ return rexBuilder.makeLiteral(literal.booleanValue());
+ }
+ }
+ }
}
diff --git a/core/src/main/java/org/apache/calcite/util/Holder.java
b/core/src/main/java/org/apache/calcite/util/Holder.java
index bc5567e8c0..78aedcd3ee 100644
--- a/core/src/main/java/org/apache/calcite/util/Holder.java
+++ b/core/src/main/java/org/apache/calcite/util/Holder.java
@@ -64,6 +64,6 @@ public static <E> Holder<E> of(E e) {
/** Creates a holder containing null. */
@SuppressWarnings("ConstantConditions")
public static <E> Holder<@Nullable E> empty() {
- return new Holder<@Nullable E>(null);
+ return new Holder<>(null);
}
}
diff --git a/core/src/main/java/org/apache/calcite/util/mapping/Mappings.java
b/core/src/main/java/org/apache/calcite/util/mapping/Mappings.java
index 82a3c74039..0139473bc5 100644
--- a/core/src/main/java/org/apache/calcite/util/mapping/Mappings.java
+++ b/core/src/main/java/org/apache/calcite/util/mapping/Mappings.java
@@ -319,16 +319,25 @@ public static <T> List<T> permute(final List<T> list,
*/
@CheckReturnValue
public static List<@Nullable Integer> asList(final TargetMapping mapping) {
- return new AbstractList<@Nullable Integer>() {
- @Override public @Nullable Integer get(int source) {
- int target = mapping.getTargetOpt(source);
- return target < 0 ? null : target;
- }
+ return new MappingsList(mapping);
+ }
- @Override public int size() {
- return mapping.getSourceCount();
- }
- };
+ /** List view of a {@link TargetMapping}. */
+ private static class MappingsList extends AbstractList<@Nullable Integer> {
+ private final TargetMapping mapping;
+
+ MappingsList(TargetMapping mapping) {
+ this.mapping = mapping;
+ }
+
+ @Override public @Nullable Integer get(int source) {
+ int target = mapping.getTargetOpt(source);
+ return target < 0 ? null : target;
+ }
+
+ @Override public int size() {
+ return mapping.getSourceCount();
+ }
}
/**
diff --git
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvFilterableTable.java
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvFilterableTable.java
index ee621740f9..212aff0ce5 100644
---
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvFilterableTable.java
+++
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvFilterableTable.java
@@ -63,12 +63,28 @@ public CsvFilterableTable(Source source,
filters.removeIf(filter -> addFilter(filter, filterValues));
final List<Integer> fields = ImmutableIntList.identity(fieldTypes.size());
final AtomicBoolean cancelFlag =
DataContext.Variable.CANCEL_FLAG.get(root);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new CsvEnumerator<>(source, cancelFlag, false, filterValues,
- CsvEnumerator.arrayConverter(fieldTypes, fields, false), ',');
- }
- };
+ return new CsvFilterableTableEnumerable(cancelFlag, filterValues,
fieldTypes, fields);
+ }
+
+ /** Enumerable for {@link CsvFilterableTable}. */
+ private class CsvFilterableTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final AtomicBoolean cancelFlag;
+ private final @Nullable String[] filterValues;
+ private final List<RelDataType> fieldTypes;
+ private final List<Integer> fields;
+
+ CsvFilterableTableEnumerable(AtomicBoolean cancelFlag, @Nullable String[]
filterValues,
+ List<RelDataType> fieldTypes, List<Integer> fields) {
+ this.cancelFlag = cancelFlag;
+ this.filterValues = filterValues;
+ this.fieldTypes = fieldTypes;
+ this.fields = fields;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new CsvEnumerator<>(source, cancelFlag, false, filterValues,
+ CsvEnumerator.arrayConverter(fieldTypes, fields, false), ',');
+ }
}
private static boolean addFilter(RexNode filter, @Nullable Object[]
filterValues) {
diff --git
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvScannableTable.java
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvScannableTable.java
index 836af81373..046825bd35 100644
---
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvScannableTable.java
+++
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvScannableTable.java
@@ -55,11 +55,25 @@ public class CsvScannableTable extends CsvTable
final List<RelDataType> fieldTypes = getFieldTypes(typeFactory);
final List<Integer> fields = ImmutableIntList.identity(fieldTypes.size());
final AtomicBoolean cancelFlag =
DataContext.Variable.CANCEL_FLAG.get(root);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new CsvEnumerator<>(source, cancelFlag, false, null,
- CsvEnumerator.arrayConverter(fieldTypes, fields, false), ',');
- }
- };
+ return new CsvScannableTableEnumerable(cancelFlag, fieldTypes, fields);
+ }
+
+ /** Enumerable for {@link CsvScannableTable}. */
+ private class CsvScannableTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final AtomicBoolean cancelFlag;
+ private final List<RelDataType> fieldTypes;
+ private final List<Integer> fields;
+
+ CsvScannableTableEnumerable(AtomicBoolean cancelFlag, List<RelDataType>
fieldTypes,
+ List<Integer> fields) {
+ this.cancelFlag = cancelFlag;
+ this.fieldTypes = fieldTypes;
+ this.fields = fields;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new CsvEnumerator<>(source, cancelFlag, false, null,
+ CsvEnumerator.arrayConverter(fieldTypes, fields, false), ',');
+ }
}
}
diff --git
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvStreamScannableTable.java
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvStreamScannableTable.java
index 7c0d574cc7..98ab6a03bc 100644
---
a/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvStreamScannableTable.java
+++
b/example/csv/src/main/java/org/apache/calcite/adapter/csv/CsvStreamScannableTable.java
@@ -62,12 +62,26 @@ public class CsvStreamScannableTable extends
CsvScannableTable
final List<RelDataType> fieldTypes = getFieldTypes(typeFactory);
final List<Integer> fields = ImmutableIntList.identity(fieldTypes.size());
final AtomicBoolean cancelFlag =
DataContext.Variable.CANCEL_FLAG.get(root);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new CsvEnumerator<>(source, cancelFlag, true, null,
- CsvEnumerator.arrayConverter(fieldTypes, fields, true), ',');
- }
- };
+ return new CsvStreamScannableTableEnumerable(cancelFlag, fieldTypes,
fields);
+ }
+
+ /** Enumerable for {@link CsvStreamScannableTable}. */
+ private class CsvStreamScannableTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final AtomicBoolean cancelFlag;
+ private final List<RelDataType> fieldTypes;
+ private final List<Integer> fields;
+
+ CsvStreamScannableTableEnumerable(AtomicBoolean cancelFlag,
List<RelDataType> fieldTypes,
+ List<Integer> fields) {
+ this.cancelFlag = cancelFlag;
+ this.fieldTypes = fieldTypes;
+ this.fields = fields;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new CsvEnumerator<>(source, cancelFlag, true, null,
+ CsvEnumerator.arrayConverter(fieldTypes, fields, true), ',');
+ }
}
@Override public Table stream() {
diff --git
a/example/function/src/main/java/org/apache/calcite/example/maze/MazeTable.java
b/example/function/src/main/java/org/apache/calcite/example/maze/MazeTable.java
index 0818cc0a74..1068bf25cb 100644
---
a/example/function/src/main/java/org/apache/calcite/example/maze/MazeTable.java
+++
b/example/function/src/main/java/org/apache/calcite/example/maze/MazeTable.java
@@ -95,17 +95,26 @@ public static ScannableTable solve(int width, int height,
int seed) {
if (Maze.DEBUG) {
maze.print(pw, true);
}
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Set<Integer> solutionSet;
- if (solution) {
- solutionSet = maze.solve(0, 0);
- } else {
- solutionSet = null;
- }
- return Linq4j.transform(maze.enumerator(solutionSet),
- s -> new Object[] {s});
+ return new MazeTableEnumerable(maze);
+ }
+
+ /** Enumerable for {@link MazeTable}. */
+ private class MazeTableEnumerable extends AbstractEnumerable<@Nullable
Object[]> {
+ private final Maze maze;
+
+ MazeTableEnumerable(Maze maze) {
+ this.maze = maze;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Set<Integer> solutionSet;
+ if (solution) {
+ solutionSet = maze.solve(0, 0);
+ } else {
+ solutionSet = null;
}
- };
+ return Linq4j.transform(maze.enumerator(solutionSet),
+ s -> new Object[] {s});
+ }
}
}
diff --git
a/file/src/main/java/org/apache/calcite/adapter/file/JsonScannableTable.java
b/file/src/main/java/org/apache/calcite/adapter/file/JsonScannableTable.java
index 8f9d2bde29..f0f2373963 100644
--- a/file/src/main/java/org/apache/calcite/adapter/file/JsonScannableTable.java
+++ b/file/src/main/java/org/apache/calcite/adapter/file/JsonScannableTable.java
@@ -46,11 +46,20 @@ public JsonScannableTable(Source source) {
}
@Override public Enumerable<@Nullable Object[]> scan(DataContext root) {
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- JavaTypeFactory typeFactory = root.getTypeFactory();
- return new JsonEnumerator(getDataList(typeFactory));
- }
- };
+ return new JsonScannableTableEnumerable(root);
+ }
+
+ /** Enumerable for {@link JsonScannableTable}. */
+ private class JsonScannableTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final DataContext root;
+
+ JsonScannableTableEnumerable(DataContext root) {
+ this.root = root;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ JavaTypeFactory typeFactory = root.getTypeFactory();
+ return new JsonEnumerator(getDataList(typeFactory));
+ }
}
}
diff --git
a/geode/src/main/java/org/apache/calcite/adapter/geode/simple/GeodeSimpleScannableTable.java
b/geode/src/main/java/org/apache/calcite/adapter/geode/simple/GeodeSimpleScannableTable.java
index f48698c7c6..1594b30e27 100644
---
a/geode/src/main/java/org/apache/calcite/adapter/geode/simple/GeodeSimpleScannableTable.java
+++
b/geode/src/main/java/org/apache/calcite/adapter/geode/simple/GeodeSimpleScannableTable.java
@@ -58,18 +58,29 @@ public GeodeSimpleScannableTable(String regionName,
RelDataType relDataType,
}
@Override public Enumerable<@Nullable Object[]> scan(DataContext root) {
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- return new GeodeSimpleEnumerator<@Nullable Object[]>(clientCache,
regionName) {
- @Override public @Nullable Object[] convert(Object obj) {
- Object values = convertToRowValues(relDataType.getFieldList(),
obj);
- if (values instanceof Object[]) {
- return (Object[]) values;
- }
- return new Object[]{values};
- }
- };
+ return new GeodeSimpleScannableTableEnumerable();
+ }
+
+ /** Enumerable for {@link GeodeSimpleScannableTable}. */
+ private class GeodeSimpleScannableTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ return new GeodeSimpleScannableTableEnumerator();
+ }
+ }
+
+ /** Enumerator for {@link GeodeSimpleScannableTable}. */
+ private class GeodeSimpleScannableTableEnumerator
+ extends GeodeSimpleEnumerator<@Nullable Object[]> {
+ GeodeSimpleScannableTableEnumerator() {
+ super(GeodeSimpleScannableTable.this.clientCache,
GeodeSimpleScannableTable.this.regionName);
+ }
+
+ @Override public @Nullable Object[] convert(Object obj) {
+ Object values = convertToRowValues(relDataType.getFieldList(), obj);
+ if (values instanceof Object[]) {
+ return (Object[]) values;
}
- };
+ return new Object[]{values};
+ }
}
}
diff --git
a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java
b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java
index e432eb8039..f73941c941 100644
--- a/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java
+++ b/kafka/src/main/java/org/apache/calcite/adapter/kafka/KafkaStreamTable.java
@@ -60,31 +60,40 @@ public class KafkaStreamTable implements ScannableTable,
StreamableTable {
@Override public Enumerable<@Nullable Object[]> scan(final DataContext root)
{
final AtomicBoolean cancelFlag =
DataContext.Variable.CANCEL_FLAG.get(root);
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- if (tableOptions.getConsumer() != null) {
- return new KafkaMessageEnumerator(tableOptions.getConsumer(),
- tableOptions.getRowConverter(), cancelFlag);
- }
-
- Properties consumerConfig = new Properties();
- consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
- tableOptions.getBootstrapServers());
- // by default it's <byte[], byte[]>
- consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.ByteArrayDeserializer");
- consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
- "org.apache.kafka.common.serialization.ByteArrayDeserializer");
-
- if (tableOptions.getConsumerParams() != null) {
- consumerConfig.putAll(tableOptions.getConsumerParams());
- }
- Consumer consumer = new KafkaConsumer<>(consumerConfig);
-
consumer.subscribe(Collections.singletonList(tableOptions.getTopicName()));
-
- return new KafkaMessageEnumerator(consumer,
tableOptions.getRowConverter(), cancelFlag);
+ return new KafkaStreamTableEnumerable(cancelFlag);
+ }
+
+ /** Enumerable for {@link KafkaStreamTable}. */
+ private class KafkaStreamTableEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final AtomicBoolean cancelFlag;
+
+ KafkaStreamTableEnumerable(AtomicBoolean cancelFlag) {
+ this.cancelFlag = cancelFlag;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ if (tableOptions.getConsumer() != null) {
+ return new KafkaMessageEnumerator(tableOptions.getConsumer(),
+ tableOptions.getRowConverter(), cancelFlag);
}
- };
+
+ Properties consumerConfig = new Properties();
+ consumerConfig.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
+ tableOptions.getBootstrapServers());
+ // by default it's <byte[], byte[]>
+ consumerConfig.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+ consumerConfig.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
+ "org.apache.kafka.common.serialization.ByteArrayDeserializer");
+
+ if (tableOptions.getConsumerParams() != null) {
+ consumerConfig.putAll(tableOptions.getConsumerParams());
+ }
+ Consumer consumer = new KafkaConsumer<>(consumerConfig);
+
consumer.subscribe(Collections.singletonList(tableOptions.getTopicName()));
+
+ return new KafkaMessageEnumerator(consumer,
tableOptions.getRowConverter(), cancelFlag);
+ }
}
@Override public RelDataType getRowType(final RelDataTypeFactory
typeFactory) {
diff --git
a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
index 0ff4afddb6..0eca4b6d52 100644
--- a/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
+++ b/linq4j/src/main/java/org/apache/calcite/linq4j/EnumerableDefaults.java
@@ -4879,32 +4879,35 @@ protected WrapMap(Function0<Map<Wrapped<K>, V>>
mapProvider, EqualityComparer<K>
}
@Override public Set<Entry<@KeyFor("this") K, V>> entrySet() {
- return new AbstractSet<Entry<@KeyFor("this") K, V>>() {
- @SuppressWarnings("override.return.invalid")
- @Override public Iterator<Entry<K, V>> iterator() {
- final Iterator<Entry<Wrapped<K>, V>> iterator =
- map.entrySet().iterator();
-
- return new Iterator<Entry<K, V>>() {
- @Override public boolean hasNext() {
- return iterator.hasNext();
- }
+ return new WrapMapEntrySet();
+ }
- @Override public Entry<K, V> next() {
- Entry<Wrapped<K>, V> next = iterator.next();
- return new SimpleEntry<>(next.getKey().element, next.getValue());
- }
+ /** EntrySet for {@link WrapMap}. */
+ private class WrapMapEntrySet extends AbstractSet<Entry<@KeyFor("this") K,
V>> {
+ @SuppressWarnings("override.return.invalid")
+ @Override public Iterator<Entry<K, V>> iterator() {
+ final Iterator<Entry<Wrapped<K>, V>> iterator =
+ map.entrySet().iterator();
- @Override public void remove() {
- iterator.remove();
- }
- };
- }
+ return new Iterator<Entry<K, V>>() {
+ @Override public boolean hasNext() {
+ return iterator.hasNext();
+ }
- @Override public int size() {
- return map.size();
- }
- };
+ @Override public Entry<K, V> next() {
+ Entry<Wrapped<K>, V> next = iterator.next();
+ return new SimpleEntry<>(next.getKey().element, next.getValue());
+ }
+
+ @Override public void remove() {
+ iterator.remove();
+ }
+ };
+ }
+
+ @Override public int size() {
+ return map.size();
+ }
}
@SuppressWarnings("contracts.conditional.postcondition.not.satisfied")
diff --git
a/plus/src/main/java/org/apache/calcite/adapter/os/FilesTableFunction.java
b/plus/src/main/java/org/apache/calcite/adapter/os/FilesTableFunction.java
index 3df9498c35..cd7ca9e8b8 100644
--- a/plus/src/main/java/org/apache/calcite/adapter/os/FilesTableFunction.java
+++ b/plus/src/main/java/org/apache/calcite/adapter/os/FilesTableFunction.java
@@ -217,121 +217,148 @@ private boolean isGnuStat() {
default:
enumerable = sourceLinux();
}
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Enumerator<String> e = enumerable.enumerator();
- return new Enumerator<@Nullable Object[]>() {
- @Nullable Object @Nullable [] current;
+ return new FilesTableFunctionEnumerable(enumerable, fieldNames,
osName);
+ }
+ };
+ }
- @Override public Object[] current() {
- return requireNonNull(current, "current");
- }
+ /** Enumerable for {@link FilesTableFunction}. */
+ private static class FilesTableFunctionEnumerable extends
AbstractEnumerable<@Nullable Object[]> {
+ private final Enumerable<String> enumerable;
+ private final List<String> fieldNames;
+ private final String osName;
- @Override public boolean moveNext() {
- current = new Object[fieldNames.size()];
- for (int i = 0; i < current.length; i++) {
- if (!e.moveNext()) {
- return false;
- }
- final String v = e.current();
- try {
- current[i] = field(fieldNames.get(i), v);
- } catch (RuntimeException e) {
- throw new RuntimeException("while parsing value ["
- + v + "] of field [" + fieldNames.get(i)
- + "] in line [" + Arrays.toString(current) + "]", e);
- }
- }
- switch (osName) {
- case "Mac OS X":
- // post-process fields: compute filename, dir_name, depth
from path
- String path = requireNonNull((String) current[14]);
- if (".".equals(path)) {
- current[14] = path = "";
- current[3] = 0; // depth
- } else if (path.startsWith("./")) {
- current[14] = path = path.substring(2);
- current[3] = count(path, '/') + 1; // depth
- } else {
- current[3] = count(path, '/'); // depth
- }
- final int slash = path.lastIndexOf('/');
- if (slash >= 0) {
- current[5] = path.substring(slash + 1); // filename
- current[9] = path.substring(0, slash); // dir_name
- } else {
- current[5] = path; // filename
- current[9] = ""; // dir_name
- }
+ FilesTableFunctionEnumerable(Enumerable<String> enumerable, List<String>
fieldNames,
+ String osName) {
+ this.enumerable = enumerable;
+ this.fieldNames = fieldNames;
+ this.osName = osName;
+ }
- // detect output format: BSD outputs single chars, GNU
outputs words
- final String type = requireNonNull((String) current[19]);
- if (type.length() > 1) {
- // GNU stat outputs descriptive types like "regular file",
"directory"
- current[19] = type.contains("directory") ? "d"
- : type.contains("regular") ? "f"
- : type.contains("symbolic") ? "l"
- : type.contains("block") ? "b"
- : type.contains("character") ? "c"
- : type.contains("fifo") ? "p"
- : type.contains("socket") ? "s"
- : "?";
- } else {
- // BSD stat outputs single characters: "/" "*" "@" or ""
- current[19] = "/".equals(type) ? "d"
- : "".equals(type) || "*".equals(type) ? "f"
- : "@".equals(type) ? "l"
- : type;
- }
- break;
- default:
- break;
- }
- return true;
- }
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Enumerator<String> e = enumerable.enumerator();
+ return new FilesTableFunctionEnumerator(fieldNames, e, osName);
+ }
+ }
- private int count(String s, char c) {
- int n = 0;
- for (int i = 0, len = s.length(); i < len; i++) {
- if (s.charAt(i) == c) {
- ++n;
- }
- }
- return n;
- }
+ /** Enumerator for {@link FilesTableFunction}. */
+ private static class FilesTableFunctionEnumerator implements
Enumerator<@Nullable Object[]> {
+ private final List<String> fieldNames;
+ private final Enumerator<String> e;
+ private final String osName;
+ @Nullable Object @Nullable [] current;
- @Override public void reset() {
- throw new UnsupportedOperationException();
- }
+ FilesTableFunctionEnumerator(List<String> fieldNames, Enumerator<String> e,
+ String osName) {
+ this.fieldNames = fieldNames;
+ this.e = e;
+ this.osName = osName;
+ }
- @Override public void close() {
- e.close();
- }
+ @Override public Object[] current() {
+ return requireNonNull(current, "current");
+ }
- private Object field(String field, String value) {
- switch (field) {
- case "block_count":
- case "depth":
- case "device":
- case "gid":
- case "uid":
- case "hard":
- return Integer.valueOf(value);
- case "inode":
- case "size":
- return Long.valueOf(value);
- case "access_time":
- case "change_time":
- case "mod_time":
- return new BigDecimal(value).multiply(THOUSAND).longValue();
- default:
- return value;
- }
- }
- };
- }
- };
+ @Override public boolean moveNext() {
+ current = new Object[fieldNames.size()];
+ for (int i = 0; i < current.length; i++) {
+ if (!e.moveNext()) {
+ return false;
+ }
+ final String v = e.current();
+ try {
+ current[i] = field(fieldNames.get(i), v);
+ } catch (RuntimeException e) {
+ throw new RuntimeException("while parsing value ["
+ + v + "] of field [" + fieldNames.get(i)
+ + "] in line [" + Arrays.toString(current) + "]", e);
+ }
}
- };
+ switch (osName) {
+ case "Mac OS X":
+ // post-process fields: compute filename, dir_name, depth from path
+ String path = requireNonNull((String) current[14]);
+ if (".".equals(path)) {
+ current[14] = path = "";
+ current[3] = 0; // depth
+ } else if (path.startsWith("./")) {
+ current[14] = path = path.substring(2);
+ current[3] = count(path, '/') + 1; // depth
+ } else {
+ current[3] = count(path, '/'); // depth
+ }
+ final int slash = path.lastIndexOf('/');
+ if (slash >= 0) {
+ current[5] = path.substring(slash + 1); // filename
+ current[9] = path.substring(0, slash); // dir_name
+ } else {
+ current[5] = path; // filename
+ current[9] = ""; // dir_name
+ }
+
+ // detect output format: BSD outputs single chars, GNU outputs words
+ final String type = requireNonNull((String) current[19]);
+ if (type.length() > 1) {
+ // GNU stat outputs descriptive types like "regular file",
"directory"
+ current[19] = type.contains("directory") ? "d"
+ : type.contains("regular") ? "f"
+ : type.contains("symbolic") ? "l"
+ : type.contains("block") ? "b"
+ : type.contains("character") ? "c"
+ : type.contains("fifo") ? "p"
+ : type.contains("socket") ? "s"
+ : "?";
+ } else {
+ // BSD stat outputs single characters: "/" "*" "@" or ""
+ current[19] = "/".equals(type) ? "d"
+ : "".equals(type) || "*".equals(type) ? "f"
+ : "@".equals(type) ? "l"
+ : type;
+ }
+ break;
+ default:
+ break;
+ }
+ return true;
+ }
+
+ private int count(String s, char c) {
+ int n = 0;
+ for (int i = 0, len = s.length(); i < len; i++) {
+ if (s.charAt(i) == c) {
+ ++n;
+ }
+ }
+ return n;
+ }
+
+ @Override public void reset() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public void close() {
+ e.close();
+ }
+
+ private Object field(String field, String value) {
+ switch (field) {
+ case "block_count":
+ case "depth":
+ case "device":
+ case "gid":
+ case "uid":
+ case "hard":
+ return Integer.valueOf(value);
+ case "inode":
+ case "size":
+ return Long.valueOf(value);
+ case "access_time":
+ case "change_time":
+ case "mod_time":
+ return new BigDecimal(value).multiply(THOUSAND).longValue();
+ default:
+ return value;
+ }
+ }
}
}
diff --git
a/plus/src/main/java/org/apache/calcite/adapter/os/GitCommitsTableFunction.java
b/plus/src/main/java/org/apache/calcite/adapter/os/GitCommitsTableFunction.java
index 0f31e6763c..2e58957291 100644
---
a/plus/src/main/java/org/apache/calcite/adapter/os/GitCommitsTableFunction.java
+++
b/plus/src/main/java/org/apache/calcite/adapter/os/GitCommitsTableFunction.java
@@ -51,91 +51,110 @@ public static ScannableTable eval(boolean b) {
@Override public Enumerable<@Nullable Object[]> scan(DataContext root) {
final Enumerable<String> enumerable =
Processes.processLines("git", "log", "--pretty=raw");
- return new AbstractEnumerable<@Nullable Object[]>() {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Enumerator<String> e = enumerable.enumerator();
- return new Enumerator<@Nullable Object[]>() {
- private @Nullable Object @Nullable [] objects;
- private final StringBuilder b = new StringBuilder();
-
- @Override public @Nullable Object[] current() {
- if (objects == null) {
- throw new NoSuchElementException();
- }
- return objects;
- }
+ return new GitCommitsTableFunctionEnumerable(enumerable);
+ }
- @Override public boolean moveNext() {
- if (!e.moveNext()) {
- objects = null;
- return false;
- }
- objects = new Object[9];
- for (;;) {
- final String line = e.current();
- if (line.isEmpty()) {
- break; // next line will be start of comments
- }
- if (line.startsWith("commit ")) {
- objects[0] = line.substring("commit ".length());
- } else if (line.startsWith("tree ")) {
- objects[1] = line.substring("tree ".length());
- } else if (line.startsWith("parent ")) {
- if (objects[2] == null) {
- objects[2] = line.substring("parent ".length());
- } else {
- objects[3] = line.substring("parent ".length());
- }
- } else if (line.startsWith("author ")) {
- objects[4] =
- line.substring("author ".length(),
- line.length() - TS_OFF.length() - 1);
- objects[5] =
- parseLong(
- line.substring(line.length() - TS_OFF.length(),
- line.length() - OFF.length() - 1)) * 1000;
- } else if (line.startsWith("committer ")) {
- objects[6] =
- line.substring("committer ".length(),
- line.length() - TS_OFF.length() - 1);
- objects[7] =
- parseLong(
- line.substring(line.length() - TS_OFF.length(),
- line.length() - OFF.length() - 1)) * 1000;
- }
- if (!e.moveNext()) {
- // We have a row, and it's the last because input is empty
- return true;
- }
- }
- for (;;) {
- if (!e.moveNext()) {
- // We have a row, and it's the last because input is empty
- objects[8] = b.toString();
- b.setLength(0);
- return true;
- }
- final String line = e.current();
- if (line.isEmpty()) {
- // We're seeing the empty line at the end of message
- objects[8] = b.toString();
- b.setLength(0);
- return true;
- }
- b.append(line.substring(" ".length())).append("\n");
- }
- }
+ /** Enumerable for {@link GitCommitsTableFunction}. */
+ class GitCommitsTableFunctionEnumerable
+ extends AbstractEnumerable<@Nullable Object[]> {
+ private final Enumerable<String> enumerable;
- @Override public void reset() {
- throw new UnsupportedOperationException();
- }
+ GitCommitsTableFunctionEnumerable(Enumerable<String> enumerable) {
+ this.enumerable = enumerable;
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Enumerator<String> e = enumerable.enumerator();
+ return new GitCommitsTableFunctionEnumerator(e);
+ }
+ }
+
+ /** Enumerator for {@link GitCommitsTableFunction}. */
+ class GitCommitsTableFunctionEnumerator implements Enumerator<@Nullable
Object[]> {
+ private final Enumerator<String> e;
+ private @Nullable Object @Nullable [] objects;
+ private final StringBuilder b;
- @Override public void close() {
- e.close();
+ GitCommitsTableFunctionEnumerator(Enumerator<String> e) {
+ this.e = e;
+ b = new StringBuilder();
+ }
+
+ @Override public @Nullable Object[] current() {
+ if (objects == null) {
+ throw new NoSuchElementException();
+ }
+ return objects;
+ }
+
+ @Override public boolean moveNext() {
+ if (!e.moveNext()) {
+ objects = null;
+ return false;
+ }
+ objects = new Object[9];
+ for (;;) {
+ final String line = e.current();
+ if (line.isEmpty()) {
+ break; // next line will be start of comments
+ }
+ if (line.startsWith("commit ")) {
+ objects[0] = line.substring("commit ".length());
+ } else if (line.startsWith("tree ")) {
+ objects[1] = line.substring("tree ".length());
+ } else if (line.startsWith("parent ")) {
+ if (objects[2] == null) {
+ objects[2] = line.substring("parent ".length());
+ } else {
+ objects[3] = line.substring("parent ".length());
}
- };
+ } else if (line.startsWith("author ")) {
+ objects[4] =
+ line.substring("author ".length(),
+ line.length() - TS_OFF.length() - 1);
+ objects[5] =
+ parseLong(
+ line.substring(line.length() - TS_OFF.length(),
+ line.length() - OFF.length() - 1)) * 1000;
+ } else if (line.startsWith("committer ")) {
+ objects[6] =
+ line.substring("committer ".length(),
+ line.length() - TS_OFF.length() - 1);
+ objects[7] =
+ parseLong(
+ line.substring(line.length() - TS_OFF.length(),
+ line.length() - OFF.length() - 1)) * 1000;
+ }
+ if (!e.moveNext()) {
+ // We have a row, and it's the last because input is empty
+ return true;
+ }
}
- };
+ for (;;) {
+ if (!e.moveNext()) {
+ // We have a row, and it's the last because input is empty
+ objects[8] = b.toString();
+ b.setLength(0);
+ return true;
+ }
+ final String line = e.current();
+ if (line.isEmpty()) {
+ // We're seeing the empty line at the end of message
+ objects[8] = b.toString();
+ b.setLength(0);
+ return true;
+ }
+ b.append(line.substring(" ".length())).append("\n");
+ }
+ }
+
+ @Override public void reset() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override public void close() {
+ e.close();
+ }
}
@Override public RelDataType getRowType(RelDataTypeFactory typeFactory) {
diff --git
a/plus/src/main/java/org/apache/calcite/adapter/tpcds/TpcdsSchema.java
b/plus/src/main/java/org/apache/calcite/adapter/tpcds/TpcdsSchema.java
index 548c20f8c2..20ae7d3145 100644
--- a/plus/src/main/java/org/apache/calcite/adapter/tpcds/TpcdsSchema.java
+++ b/plus/src/main/java/org/apache/calcite/adapter/tpcds/TpcdsSchema.java
@@ -158,36 +158,45 @@ private class TpcdsQueryableTable<E extends
com.teradata.tpcds.Table>
@Override public <T> Queryable<T> asQueryable(final QueryProvider
queryProvider,
final SchemaPlus schema, final String tableName) {
//noinspection unchecked
- return (Queryable) new AbstractTableQueryable<@Nullable
Object[]>(queryProvider,
- schema, this, tableName) {
- @Override public Enumerator<@Nullable Object[]> enumerator() {
- final Session session =
- Session.getDefaultSession()
- .withTable(tpcdsTable)
- .withScale(scaleFactor);
- final Results results = Results.constructResults(tpcdsTable,
session);
- return Linq4j.asEnumerable(results)
- .selectMany(
- new Function1<List<List<@Nullable String>>,
Enumerable<@Nullable Object[]>>() {
- final Column[] columns = tpcdsTable.getColumns();
-
- @Override public Enumerable<@Nullable Object[]> apply(
- List<List<@Nullable String>> inRows) {
- final List<@Nullable Object[]> rows = new ArrayList<>();
- for (List<@Nullable String> strings : inRows) {
- final @Nullable Object[] values = new
Object[columns.length];
- for (int i = 0; i < strings.size(); i++) {
- values[i] = convert(strings.get(i), columns[i]);
- }
- rows.add(values);
- }
- return Linq4j.asEnumerable(rows);
- }
-
- })
- .enumerator();
+ return (Queryable) new TpcdsSchemaQueryable(queryProvider, schema,
tableName);
+ }
+
+ /** Queryable for {@link TpcdsSchema}. */
+ private class TpcdsSchemaQueryable extends
AbstractTableQueryable<@Nullable Object[]> {
+ TpcdsSchemaQueryable(QueryProvider queryProvider, SchemaPlus schema,
+ String tableName) {
+ super(queryProvider, schema, TpcdsQueryableTable.this, tableName);
+ }
+
+ @Override public Enumerator<@Nullable Object[]> enumerator() {
+ final Session session =
+ Session.getDefaultSession()
+ .withTable(tpcdsTable)
+ .withScale(scaleFactor);
+ final Results results = Results.constructResults(tpcdsTable, session);
+ return Linq4j.asEnumerable(results)
+ .selectMany(new TpcdsSchemaSelector())
+ .enumerator();
+ }
+ }
+
+ /** Selector for {@link TpcdsSchema}. */
+ private class TpcdsSchemaSelector
+ implements Function1<List<List<@Nullable String>>,
Enumerable<@Nullable Object[]>> {
+ final Column[] columns = tpcdsTable.getColumns();
+
+ @Override public Enumerable<@Nullable Object[]> apply(
+ List<List<@Nullable String>> inRows) {
+ final List<@Nullable Object[]> rows = new ArrayList<>();
+ for (List<@Nullable String> strings : inRows) {
+ final @Nullable Object[] values = new Object[columns.length];
+ for (int i = 0; i < strings.size(); i++) {
+ values[i] = convert(strings.get(i), columns[i]);
+ }
+ rows.add(values);
}
- };
+ return Linq4j.asEnumerable(rows);
+ }
}
@Override public RelDataType getRowType(RelDataTypeFactory typeFactory) {
