Skip to content
Merged
Original file line number Diff line number Diff line change
@@ -0,0 +1,88 @@
/*
* 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.iceberg.data;

import java.lang.reflect.Array;
import java.nio.ByteBuffer;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.time.OffsetDateTime;
import java.util.function.Function;
import org.apache.iceberg.StructLike;
import org.apache.iceberg.types.Type;
import org.apache.iceberg.types.Types;

class InternalRecordWrapper implements StructLike {
private final Function<Object, Object>[] transforms;
private StructLike wrapped = null;

@SuppressWarnings("unchecked")
InternalRecordWrapper(Types.StructType struct) {
this.transforms = struct.fields().stream()
.map(field -> converter(field.type()))
.toArray(length -> (Function<Object, Object>[]) Array.newInstance(Function.class, length));
}

private static Function<Object, Object> converter(Type type) {
switch (type.typeId()) {
case DATE:
return date -> DateTimeUtil.daysFromDate((LocalDate) date);
case TIME:
return time -> DateTimeUtil.microsFromTime((LocalTime) time);
case TIMESTAMP:
if (((Types.TimestampType) type).shouldAdjustToUTC()) {
return timestamp -> DateTimeUtil.microsFromTimestamptz((OffsetDateTime) timestamp);
} else {
return timestamp -> DateTimeUtil.microsFromTimestamp((LocalDateTime) timestamp);
}
case FIXED:
return bytes -> ByteBuffer.wrap((byte[]) bytes);
case STRUCT:
InternalRecordWrapper wrapper = new InternalRecordWrapper(type.asStructType());
return struct -> wrapper.wrap((StructLike) struct);
default:
}
return null;
}

public InternalRecordWrapper wrap(StructLike record) {
this.wrapped = record;
return this;
}

@Override
public int size() {
return wrapped.size();
}

@Override
public <T> T get(int pos, Class<T> javaClass) {
if (transforms[pos] != null) {
return javaClass.cast(transforms[pos].apply(wrapped.get(pos, Object.class)));
}
return wrapped.get(pos, javaClass);
}

@Override
public <T> void set(int pos, T value) {
throw new UnsupportedOperationException("Cannot update InternalRecordWrapper");
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -161,7 +161,10 @@ public boolean hasNext() {

if (task.residual() != null && task.residual() != Expressions.alwaysTrue()) {
Evaluator filter = new Evaluator(projection.asStruct(), task.residual(), caseSensitive);
this.currentIterator = Iterables.filter(reader, filter::eval).iterator();
this.currentIterator = Iterables.filter(reader, record -> {
InternalRecordWrapper wrapperRecord = new InternalRecordWrapper(record.struct()).wrap(record);

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The wrapper can be reused so it should be moved outside the predicate. The predicate just needs to be this:

  record -> filter.eval(wrapper.wrap(record))

return filter.eval(wrapperRecord);
}).iterator();
} else {
this.currentIterator = reader.iterator();
}
Expand Down
46 changes: 43 additions & 3 deletions data/src/test/java/org/apache/iceberg/data/TestLocalScan.java
Original file line number Diff line number Diff line change
Expand Up @@ -67,6 +67,7 @@
import static com.google.common.collect.Iterables.filter;
import static com.google.common.collect.Iterables.transform;
import static org.apache.iceberg.DataFiles.fromInputFile;
import static org.apache.iceberg.expressions.Expressions.equal;
import static org.apache.iceberg.expressions.Expressions.lessThan;
import static org.apache.iceberg.expressions.Expressions.lessThanOrEqual;
import static org.apache.iceberg.hadoop.HadoopOutputFile.fromPath;
Expand Down Expand Up @@ -391,13 +392,17 @@ public void testAsOfTimeOlderThanFirstSnapshot() {
}

private DataFile writeFile(String location, String filename, List<Record> records) throws IOException {
return writeFile(location, filename, SCHEMA, records);
}

private DataFile writeFile(String location, String filename, Schema schema, List<Record> records) throws IOException {
Path path = new Path(location, filename);
FileFormat fileFormat = FileFormat.fromFileName(filename);
Preconditions.checkNotNull(fileFormat, "Cannot determine format for file: %s", filename);
switch (fileFormat) {
case AVRO:
FileAppender<Record> avroAppender = Avro.write(fromPath(path, CONF))
.schema(SCHEMA)
.schema(schema)
.createWriterFunc(DataWriter::create)
.named(fileFormat.name())
.build();
Expand All @@ -414,7 +419,7 @@ private DataFile writeFile(String location, String filename, List<Record> record

case PARQUET:
FileAppender<Record> parquetAppender = Parquet.write(fromPath(path, CONF))
.schema(SCHEMA)
.schema(schema)
.createWriterFunc(GenericParquetWriter::buildWriter)
.build();
try {
Expand All @@ -430,7 +435,7 @@ private DataFile writeFile(String location, String filename, List<Record> record

case ORC:
FileAppender<Record> orcAppender = ORC.write(fromPath(path, CONF))
.schema(SCHEMA)
.schema(schema)
.createWriterFunc(GenericOrcWriter::buildWriter)
.build();
try {
Expand All @@ -449,6 +454,41 @@ private DataFile writeFile(String location, String filename, List<Record> record
}
}

@Test
public void testFilterWithDateAndTimestamp() throws IOException {
Schema schema = new Schema(
required(1, "timestamp_with_zone", Types.TimestampType.withZone()),
required(2, "timestamp_without_zone", Types.TimestampType.withoutZone()),
required(3, "date", Types.DateType.get()),
required(4, "time", Types.TimeType.get())
);

File tableLocation = temp.newFolder("complex_filter_table");
Assert.assertTrue(tableLocation.delete());

Table table = TABLES.create(
schema, PartitionSpec.unpartitioned(),
ImmutableMap.of(TableProperties.DEFAULT_FILE_FORMAT, format.name()),
tableLocation.getAbsolutePath());

List<Record> expected = RandomGenericData.generate(schema, 100, 435691832918L);
DataFile file = writeFile(tableLocation.toString(), format.addExtension("record-file"), schema, expected);
table.newFastAppend().appendFile(file).commit();

for (Record r : expected) {
Iterable<Record> filterResult = IcebergGenerics.read(table)
.where(equal("timestamp_with_zone", r.getField("timestamp_with_zone").toString()))
.where(equal("timestamp_without_zone", r.getField("timestamp_without_zone").toString()))
.where(equal("date", r.getField("date").toString()))
.where(equal("time", r.getField("time").toString()))
.build();

Assert.assertTrue(filterResult.iterator().hasNext());
Record readRecord = filterResult.iterator().next();
Assert.assertEquals(r.getField("timestamp_with_zone"), readRecord.getField("timestamp_with_zone"));
}
}

private static ByteBuffer longToBuffer(long value) {
return ByteBuffer.allocate(8).order(ByteOrder.LITTLE_ENDIAN).putLong(0, value);
}
Expand Down