You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@druid.apache.org by "abhishekrb19 (via GitHub)" <gi...@apache.org> on 2024/01/24 17:29:25 UTC
[PR] Extension to read and ingest Delta Lake tables (druid)
abhishekrb19 opened a new pull request, #15755:
URL: https://github.com/apache/druid/pull/15755
<!-- Thanks for trying to help us make Apache Druid be the best it can be! Please fill out as much of the following information as is possible (where relevant, and remove it when irrelevant) to help make the intention and scope of this PR clear in order to ease review. -->
<!-- Please read the doc for contribution (https://github.com/apache/druid/blob/master/CONTRIBUTING.md) before making this PR. Also, once you open a PR, please _avoid using force pushes and rebasing_ since these make it difficult for reviewers to see what you've changed in response to their reviews. See [the 'If your pull request shows conflicts with master' section](https://github.com/apache/druid/blob/master/CONTRIBUTING.md#if-your-pull-request-shows-conflicts-with-master) for more details. -->
Delta Lake is an open source storage layer that brings reliability to data lakes. Users can ingest data stored in a Delta Lake table into Apache Druid via a new input source `delta`. This is available via the Delta Lake extension, add the `druid-deltalake-extensions` to the list of loaded extensions. The Delta input source reads the configured Delta Lake table and extracts all the underlying delta files in the table's latest snapshot. Delta Lake files are versioned Parquet format.
This is joint work with @abhishekagarwal87, @LakshSingla and @AmatyaAvadhanula.
TODO:
- add Kerberos-based authentication mechanism
- prune dependencies a bit
Will include a sample spec in the description once the TODOs are addressed.
#### Release note
Enhanced ingestion capabilities to support ingestion of Delta Lake tables.
This PR has:
- [x] been self-reviewed.
- [ ] using the [concurrency checklist](https://github.com/apache/druid/blob/master/dev/code-review/concurrency.md) (Remove this item if the PR doesn't have any relation to concurrency.)
- [x] added documentation for new or modified features or behaviors.
- [ x] a release note entry in the PR description.
- [x] added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
- [x] added or updated version, license, or notice information in [licenses.yaml](https://github.com/apache/druid/blob/master/dev/license.md)
- [ ] added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
- [x] added unit tests or modified existing tests to cover new code paths, ensuring the threshold for [code coverage](https://github.com/apache/druid/blob/master/dev/code-review/code-coverage.md) is met.
- [ ] added integration tests.
- [ ] been tested in a test Druid cluster.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1466773437
##########
docs/ingestion/input-sources.md:
##########
@@ -715,6 +715,13 @@ rolled-up datasource `wikipedia_rollup` by grouping on hour, "countryName", and
to `true` to enable a compatibility mode where the timestampSpec is ignored.
:::
+The [secondary partitioning method](native-batch.md#partitionsspec) determines the requisite number of concurrent worker tasks that run in parallel to complete ingestion with the Combining input source.
+Set this value in `maxNumConcurrentSubTasks` in `tuningConfig` based on the secondary partitioning method:
+- `range` or `single_dim` partitioning: greater than or equal to 1
+- `hashed` or `dynamic` partitioning: greater than or equal to 2
+
+For more information on the `maxNumConcurrentSubTasks` field, see [Implementation considerations](native-batch.md#implementation-considerations).
Review Comment:
Yeah, not directly related to this extension, but I noticed that this text was nested inside the Iceberg section: [Iceberg Filter Object](https://druid.apache.org/docs/latest/ingestion/input-sources#iceberg-filter-object) and seemed irrelevant. So I moved it to the Druid Input source, where it seems more relevant.
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
+{
+ private final io.delta.kernel.data.Row row;
+ private final StructType schema;
+ private final Object2IntMap<String> fieldNameToOrdinal = new Object2IntOpenHashMap<>();
+ private final InputRow delegateRow;
+
+ private static final ZoneId ZONE_ID = ZoneId.systemDefault();
+
+ public DeltaInputRow(io.delta.kernel.data.Row row, InputRowSchema inputRowSchema)
+ {
+ this.row = row;
+ this.schema = row.getSchema();
+ List<String> fieldNames = this.schema.fieldNames();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ fieldNameToOrdinal.put(fieldNames.get(i), i);
+ }
+ fieldNameToOrdinal.defaultReturnValue(-1);
+
+ Map<String, Object> theMap = new HashMap<>();
+ for (String fieldName : fieldNames) {
+ theMap.put(fieldName, _getRaw(fieldName));
+ }
+ delegateRow = MapInputRowParser.parse(inputRowSchema, theMap);
+ }
+
+ @Override
+ public List<String> getDimensions()
+ {
+ return delegateRow.getDimensions();
+ }
+
+ @Override
+ public long getTimestampFromEpoch()
+ {
+ return delegateRow.getTimestampFromEpoch();
+ }
+
+ @Override
+ public DateTime getTimestamp()
+ {
+ return delegateRow.getTimestamp();
+ }
+
+ @Override
+ public List<String> getDimension(String dimension)
+ {
+ return delegateRow.getDimension(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Object getRaw(String dimension)
+ {
+ return delegateRow.getRaw(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Number getMetric(String metric)
+ {
+ return delegateRow.getMetric(metric);
+ }
+
+ @Override
+ public int compareTo(Row o)
+ {
+ return this.getTimestamp().compareTo(o.getTimestamp());
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ return o instanceof DeltaInputRow && compareTo((DeltaInputRow) o) == 0;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(row, schema, fieldNameToOrdinal, delegateRow);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "DeltaInputRow{" +
+ "row=" + row +
+ ", schema=" + schema +
+ ", fieldNameToOrdinal=" + fieldNameToOrdinal +
+ ", delegateRow=" + delegateRow +
+ '}';
+ }
+
+ public Map<String, Object> getRawRowAsMap()
+ {
+ return RowSerde.convertRowToJsonObject(row);
+ }
+
+ @Nullable
+ private Object _getRaw(String dimension)
+ {
+ StructField field = schema.get(dimension);
+ if (field == null || field.isMetadataColumn()) {
+ return null;
+ }
+
+ int ordinal = fieldNameToOrdinal.getInt(dimension);
+ if (ordinal < 0) {
+ return null;
+ }
+ return getValue(field.getDataType(), row, ordinal);
+ }
+
+ @Nullable
+ private static Object getValue(DataType dataType, io.delta.kernel.data.Row dataRow, int columnOrdinal)
+ {
+ if (dataRow.isNullAt(columnOrdinal)) {
+ return null;
+ } else if (dataType instanceof BooleanType) {
+ return dataRow.getBoolean(columnOrdinal);
+ } else if (dataType instanceof ByteType) {
+ return dataRow.getByte(columnOrdinal);
+ } else if (dataType instanceof ShortType) {
+ return dataRow.getShort(columnOrdinal);
+ } else if (dataType instanceof IntegerType) {
+ return dataRow.getInt(columnOrdinal);
+ } else if (dataType instanceof DateType) {
+ // DateType data is stored internally as the number of days since 1970-01-01
+ int daysSinceEpochUTC = dataRow.getInt(columnOrdinal);
+ return LocalDate.ofEpochDay(daysSinceEpochUTC).atStartOfDay(ZONE_ID).toEpochSecond();
+ } else if (dataType instanceof LongType) {
+ return dataRow.getLong(columnOrdinal);
+ } else if (dataType instanceof TimestampType) {
+ // TimestampType data is stored internally as the number of microseconds since epoch
+ long microSecsSinceEpochUTC = dataRow.getLong(columnOrdinal);
+ LocalDateTime dateTime = LocalDateTime.ofEpochSecond(
+ microSecsSinceEpochUTC / 1_000_000 /* epochSecond */,
+ (int) (1000 * microSecsSinceEpochUTC % 1_000_000) /* nanoOfSecond */,
+ ZoneOffset.UTC
+ );
+ return dateTime.atZone(ZONE_ID).toInstant().toEpochMilli();
+ } else if (dataType instanceof FloatType) {
+ return dataRow.getFloat(columnOrdinal);
+ } else if (dataType instanceof DoubleType) {
+ return dataRow.getDouble(columnOrdinal);
+ } else if (dataType instanceof StringType) {
+ return dataRow.getString(columnOrdinal);
+ } else if (dataType instanceof BinaryType) {
+ final byte[] arr = dataRow.getBinary(columnOrdinal);
+ final char[] charArray = new char[arr.length];
+ for (int i = 0; i < arr.length; i++) {
+ charArray[i] = (char) (arr[i] & 0xff);
+ }
+ return String.valueOf(charArray);
+ } else if (dataType instanceof DecimalType) {
+ return dataRow.getDecimal(columnOrdinal).longValue();
+ } else {
+ throw InvalidInput.exception("Unsupported data type[%s]", dataType);
Review Comment:
added
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
+{
+ private final io.delta.kernel.data.Row row;
+ private final StructType schema;
+ private final Object2IntMap<String> fieldNameToOrdinal = new Object2IntOpenHashMap<>();
+ private final InputRow delegateRow;
+
+ private static final ZoneId ZONE_ID = ZoneId.systemDefault();
+
+ public DeltaInputRow(io.delta.kernel.data.Row row, InputRowSchema inputRowSchema)
+ {
+ this.row = row;
+ this.schema = row.getSchema();
+ List<String> fieldNames = this.schema.fieldNames();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ fieldNameToOrdinal.put(fieldNames.get(i), i);
+ }
+ fieldNameToOrdinal.defaultReturnValue(-1);
+
+ Map<String, Object> theMap = new HashMap<>();
+ for (String fieldName : fieldNames) {
+ theMap.put(fieldName, _getRaw(fieldName));
+ }
+ delegateRow = MapInputRowParser.parse(inputRowSchema, theMap);
+ }
+
+ @Override
+ public List<String> getDimensions()
+ {
+ return delegateRow.getDimensions();
+ }
+
+ @Override
+ public long getTimestampFromEpoch()
+ {
+ return delegateRow.getTimestampFromEpoch();
+ }
+
+ @Override
+ public DateTime getTimestamp()
+ {
+ return delegateRow.getTimestamp();
+ }
+
+ @Override
+ public List<String> getDimension(String dimension)
+ {
+ return delegateRow.getDimension(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Object getRaw(String dimension)
+ {
+ return delegateRow.getRaw(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Number getMetric(String metric)
+ {
+ return delegateRow.getMetric(metric);
+ }
+
+ @Override
+ public int compareTo(Row o)
+ {
+ return this.getTimestamp().compareTo(o.getTimestamp());
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ return o instanceof DeltaInputRow && compareTo((DeltaInputRow) o) == 0;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(row, schema, fieldNameToOrdinal, delegateRow);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "DeltaInputRow{" +
+ "row=" + row +
+ ", schema=" + schema +
+ ", fieldNameToOrdinal=" + fieldNameToOrdinal +
+ ", delegateRow=" + delegateRow +
+ '}';
+ }
+
+ public Map<String, Object> getRawRowAsMap()
+ {
+ return RowSerde.convertRowToJsonObject(row);
+ }
+
+ @Nullable
+ private Object _getRaw(String dimension)
+ {
+ StructField field = schema.get(dimension);
+ if (field == null || field.isMetadataColumn()) {
+ return null;
+ }
+
+ int ordinal = fieldNameToOrdinal.getInt(dimension);
+ if (ordinal < 0) {
+ return null;
+ }
+ return getValue(field.getDataType(), row, ordinal);
+ }
+
+ @Nullable
+ private static Object getValue(DataType dataType, io.delta.kernel.data.Row dataRow, int columnOrdinal)
+ {
+ if (dataRow.isNullAt(columnOrdinal)) {
+ return null;
+ } else if (dataType instanceof BooleanType) {
+ return dataRow.getBoolean(columnOrdinal);
+ } else if (dataType instanceof ByteType) {
+ return dataRow.getByte(columnOrdinal);
+ } else if (dataType instanceof ShortType) {
+ return dataRow.getShort(columnOrdinal);
+ } else if (dataType instanceof IntegerType) {
+ return dataRow.getInt(columnOrdinal);
+ } else if (dataType instanceof DateType) {
+ // DateType data is stored internally as the number of days since 1970-01-01
+ int daysSinceEpochUTC = dataRow.getInt(columnOrdinal);
+ return LocalDate.ofEpochDay(daysSinceEpochUTC).atStartOfDay(ZONE_ID).toEpochSecond();
+ } else if (dataType instanceof LongType) {
+ return dataRow.getLong(columnOrdinal);
+ } else if (dataType instanceof TimestampType) {
+ // TimestampType data is stored internally as the number of microseconds since epoch
+ long microSecsSinceEpochUTC = dataRow.getLong(columnOrdinal);
+ LocalDateTime dateTime = LocalDateTime.ofEpochSecond(
+ microSecsSinceEpochUTC / 1_000_000 /* epochSecond */,
+ (int) (1000 * microSecsSinceEpochUTC % 1_000_000) /* nanoOfSecond */,
+ ZoneOffset.UTC
+ );
+ return dateTime.atZone(ZONE_ID).toInstant().toEpochMilli();
+ } else if (dataType instanceof FloatType) {
+ return dataRow.getFloat(columnOrdinal);
+ } else if (dataType instanceof DoubleType) {
+ return dataRow.getDouble(columnOrdinal);
+ } else if (dataType instanceof StringType) {
+ return dataRow.getString(columnOrdinal);
+ } else if (dataType instanceof BinaryType) {
+ final byte[] arr = dataRow.getBinary(columnOrdinal);
+ final char[] charArray = new char[arr.length];
+ for (int i = 0; i < arr.length; i++) {
+ charArray[i] = (char) (arr[i] & 0xff);
+ }
+ return String.valueOf(charArray);
Review Comment:
Good catch, I was trying to convert it to an unsigned value, which is not needed in this case
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputSource.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.druid.delta.input;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Iterators;
+import com.google.common.primitives.Ints;
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.TableNotFoundException;
+import io.delta.kernel.client.TableClient;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.client.DefaultTableClient;
+import io.delta.kernel.internal.util.Utils;
+import io.delta.kernel.utils.CloseableIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.data.input.SplitHintSpec;
+import org.apache.druid.data.input.impl.SplittableInputSource;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.utils.Streams;
+import org.apache.hadoop.conf.Configuration;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Input source to ingest data from a Delta Lake.
+ * This input source reads the latest snapshot from a Delta table specified by {@code tablePath} parameter.
+ * Note: the kernel table API only supports reading from the latest snapshot.
+ */
+public class DeltaInputSource implements SplittableInputSource<DeltaSplit>
+{
+ public static final String TYPE_KEY = "delta";
+
+ @JsonProperty
+ private final String tablePath;
+
+ @JsonProperty
+ @Nullable
+ private final DeltaSplit deltaSplit;
+
+ @JsonCreator
+ public DeltaInputSource(
+ @JsonProperty("tablePath") String tablePath,
+ @JsonProperty("deltaSplit") @Nullable DeltaSplit deltaSplit
+ )
+ {
+ if (tablePath == null) {
+ throw InvalidInput.exception("tablePath cannot be null");
+ }
+ this.tablePath = tablePath;
+ this.deltaSplit = deltaSplit;
+ }
+
+ @Override
+ public boolean needsFormat()
+ {
+ // Only support Parquet
+ return false;
+ }
+
+ @Override
+ public InputSourceReader reader(
+ InputRowSchema inputRowSchema,
+ @Nullable InputFormat inputFormat,
+ File temporaryDirectory
+ )
+ {
+ Configuration hadoopConf = new Configuration();
+ TableClient tableClient = DefaultTableClient.create(hadoopConf);
+ try {
+ final Row scanState;
+ final List<Row> scanRowList;
+
Review Comment:
Added javadocs to this function
##########
extensions-contrib/druid-deltalake-extensions/src/test/resources/create_delta_table.py:
##########
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+
+# 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.
+
+import os
+
+import argparse
+import delta
+import mimesis
+import pyspark
+
+
+def config_spark_with_delta_lake():
Review Comment:
Will do. I might change this test setup a bit to get different datatype coverage, etc.
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/RowSerde.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.druid.delta.input;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.delta.kernel.client.TableClient;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.internal.data.DefaultJsonRow;
+import io.delta.kernel.internal.types.TableSchemaSerDe;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.MapType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import org.apache.druid.error.InvalidInput;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility class to serialize and deserialize {@link Row} object.
+ * Code borrowed from <a href="https://github.com/delta-io/delta/blob/master/kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/utils/RowSerDe.java">
Review Comment:
Updated, good point
##########
extensions-contrib/druid-deltalake-extensions/src/test/java/org/apache/druid/delta/input/DeltaInputSourceTest.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.druid.delta.input;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class DeltaInputSourceTest
+{
+ @Test
+ public void testReadDeltaLakeFilesSample() throws IOException
+ {
+ final DeltaInputSource deltaInputSource = new DeltaInputSource(DeltaTestUtil.DELTA_TABLE_PATH, null);
+ Assert.assertNotNull(deltaInputSource);
Review Comment:
Heh, not sure, but I removed these unnecessary assert checks
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1466773437
##########
docs/ingestion/input-sources.md:
##########
@@ -715,6 +715,13 @@ rolled-up datasource `wikipedia_rollup` by grouping on hour, "countryName", and
to `true` to enable a compatibility mode where the timestampSpec is ignored.
:::
+The [secondary partitioning method](native-batch.md#partitionsspec) determines the requisite number of concurrent worker tasks that run in parallel to complete ingestion with the Combining input source.
+Set this value in `maxNumConcurrentSubTasks` in `tuningConfig` based on the secondary partitioning method:
+- `range` or `single_dim` partitioning: greater than or equal to 1
+- `hashed` or `dynamic` partitioning: greater than or equal to 2
+
+For more information on the `maxNumConcurrentSubTasks` field, see [Implementation considerations](native-batch.md#implementation-considerations).
Review Comment:
Yeah, not directly related to this extension, but I noticed that this text was nested inside the Iceberg section: [Iceberg Filter Object](https://druid.apache.org/docs/latest/ingestion/input-sources#iceberg-filter-object) and seemed irrelevant. So I moved it to the Druid Input source, where it seems more relevant, since I was touching adjacent sections in this file.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "vkorukanti (via GitHub)" <gi...@apache.org>.
vkorukanti commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1476750537
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputSourceReader.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputStats;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+/**
+ * A reader for the Delta Lake input source. It initializes an iterator {@link DeltaInputSourceIterator}
+ * for a subset of Delta records given by {@link FilteredColumnarBatch} and schema {@link InputRowSchema}.
+ *
+ */
+public class DeltaInputSourceReader implements InputSourceReader
+{
+ private final io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator;
+ private final InputRowSchema inputRowSchema;
+
+ public DeltaInputSourceReader(
+ io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator,
+ InputRowSchema inputRowSchema
+ )
+ {
+ this.filteredColumnarBatchCloseableIterator = filteredColumnarBatchCloseableIterator;
+ this.inputRowSchema = inputRowSchema;
+ }
+
+ @Override
+ public CloseableIterator<InputRow> read()
+ {
+ return new DeltaInputSourceIterator(filteredColumnarBatchCloseableIterator, inputRowSchema);
+ }
+
+ @Override
+ public CloseableIterator<InputRow> read(InputStats inputStats)
+ {
+ return new DeltaInputSourceIterator(filteredColumnarBatchCloseableIterator, inputRowSchema);
+ }
+
+ @Override
+ public CloseableIterator<InputRowListPlusRawValues> sample()
+ {
+
+ CloseableIterator<InputRow> inner = read();
+ return new CloseableIterator<InputRowListPlusRawValues>()
+ {
+ @Override
+ public void close() throws IOException
+ {
+ inner.close();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return inner.hasNext();
+ }
+
+ @Override
+ public InputRowListPlusRawValues next()
+ {
+ DeltaInputRow deltaInputRow = (DeltaInputRow) inner.next();
+ return InputRowListPlusRawValues.of(deltaInputRow, deltaInputRow.getRawRowAsMap());
+ }
+ };
+ }
+
+ private static class DeltaInputSourceIterator implements CloseableIterator<InputRow>
+ {
+ private final io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator;
+
+ private io.delta.kernel.utils.CloseableIterator<Row> currentBatch = null;
+ private final InputRowSchema inputRowSchema;
+
+ public DeltaInputSourceIterator(
+ io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator,
+ InputRowSchema inputRowSchema
+ )
+ {
+ this.filteredColumnarBatchCloseableIterator = filteredColumnarBatchCloseableIterator;
+ this.inputRowSchema = inputRowSchema;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ while (currentBatch == null || !currentBatch.hasNext()) {
+ if (!filteredColumnarBatchCloseableIterator.hasNext()) {
+ return false; // No more batches or records to read!
+ }
+ currentBatch = filteredColumnarBatchCloseableIterator.next().getRows();
Review Comment:
one qn: do we want to check if the `currentBatch` can return `hasNext` to true? because it could contain zero rows due to the selection vector.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "vkorukanti (via GitHub)" <gi...@apache.org>.
vkorukanti commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1476750537
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputSourceReader.java:
##########
@@ -0,0 +1,138 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputStats;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+
+import java.io.IOException;
+import java.util.NoSuchElementException;
+
+/**
+ * A reader for the Delta Lake input source. It initializes an iterator {@link DeltaInputSourceIterator}
+ * for a subset of Delta records given by {@link FilteredColumnarBatch} and schema {@link InputRowSchema}.
+ *
+ */
+public class DeltaInputSourceReader implements InputSourceReader
+{
+ private final io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator;
+ private final InputRowSchema inputRowSchema;
+
+ public DeltaInputSourceReader(
+ io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator,
+ InputRowSchema inputRowSchema
+ )
+ {
+ this.filteredColumnarBatchCloseableIterator = filteredColumnarBatchCloseableIterator;
+ this.inputRowSchema = inputRowSchema;
+ }
+
+ @Override
+ public CloseableIterator<InputRow> read()
+ {
+ return new DeltaInputSourceIterator(filteredColumnarBatchCloseableIterator, inputRowSchema);
+ }
+
+ @Override
+ public CloseableIterator<InputRow> read(InputStats inputStats)
+ {
+ return new DeltaInputSourceIterator(filteredColumnarBatchCloseableIterator, inputRowSchema);
+ }
+
+ @Override
+ public CloseableIterator<InputRowListPlusRawValues> sample()
+ {
+
+ CloseableIterator<InputRow> inner = read();
+ return new CloseableIterator<InputRowListPlusRawValues>()
+ {
+ @Override
+ public void close() throws IOException
+ {
+ inner.close();
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ return inner.hasNext();
+ }
+
+ @Override
+ public InputRowListPlusRawValues next()
+ {
+ DeltaInputRow deltaInputRow = (DeltaInputRow) inner.next();
+ return InputRowListPlusRawValues.of(deltaInputRow, deltaInputRow.getRawRowAsMap());
+ }
+ };
+ }
+
+ private static class DeltaInputSourceIterator implements CloseableIterator<InputRow>
+ {
+ private final io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator;
+
+ private io.delta.kernel.utils.CloseableIterator<Row> currentBatch = null;
+ private final InputRowSchema inputRowSchema;
+
+ public DeltaInputSourceIterator(
+ io.delta.kernel.utils.CloseableIterator<FilteredColumnarBatch> filteredColumnarBatchCloseableIterator,
+ InputRowSchema inputRowSchema
+ )
+ {
+ this.filteredColumnarBatchCloseableIterator = filteredColumnarBatchCloseableIterator;
+ this.inputRowSchema = inputRowSchema;
+ }
+
+ @Override
+ public boolean hasNext()
+ {
+ while (currentBatch == null || !currentBatch.hasNext()) {
+ if (!filteredColumnarBatchCloseableIterator.hasNext()) {
+ return false; // No more batches or records to read!
+ }
+ currentBatch = filteredColumnarBatchCloseableIterator.next().getRows();
Review Comment:
one qn: do we want to check if the `currentBatch` can return `hasNext` to true? because it could contain zero rows due to the selection vector.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "abhishekrb19 (via GitHub)" <gi...@apache.org>.
abhishekrb19 commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1466773437
##########
docs/ingestion/input-sources.md:
##########
@@ -715,6 +715,13 @@ rolled-up datasource `wikipedia_rollup` by grouping on hour, "countryName", and
to `true` to enable a compatibility mode where the timestampSpec is ignored.
:::
+The [secondary partitioning method](native-batch.md#partitionsspec) determines the requisite number of concurrent worker tasks that run in parallel to complete ingestion with the Combining input source.
+Set this value in `maxNumConcurrentSubTasks` in `tuningConfig` based on the secondary partitioning method:
+- `range` or `single_dim` partitioning: greater than or equal to 1
+- `hashed` or `dynamic` partitioning: greater than or equal to 2
+
+For more information on the `maxNumConcurrentSubTasks` field, see [Implementation considerations](native-batch.md#implementation-considerations).
Review Comment:
Yeah, not directly related to this extension, but I noticed that this text was nested inside the Iceberg section: [Iceberg Filter Object](https://druid.apache.org/docs/latest/ingestion/input-sources#iceberg-filter-object) and seemed irrelevant. since I was touching adjacent sections in this file, I moved it to the Druid Input source, where it seems more relevant.
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "github-advanced-security[bot] (via GitHub)" <gi...@apache.org>.
github-advanced-security[bot] commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1466003412
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
+{
+ private final io.delta.kernel.data.Row row;
+ private final StructType schema;
+ private final Object2IntMap<String> fieldNameToOrdinal = new Object2IntOpenHashMap<>();
+ private final InputRow delegateRow;
+
+ private static final ZoneId ZONE_ID = ZoneId.systemDefault();
+
+ public DeltaInputRow(io.delta.kernel.data.Row row, InputRowSchema inputRowSchema)
+ {
+ this.row = row;
+ this.schema = row.getSchema();
+ List<String> fieldNames = this.schema.fieldNames();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ fieldNameToOrdinal.put(fieldNames.get(i), i);
+ }
+ fieldNameToOrdinal.defaultReturnValue(-1);
+
+ Map<String, Object> theMap = new HashMap<>();
+ for (String fieldName : fieldNames) {
+ theMap.put(fieldName, _getRaw(fieldName));
+ }
+ delegateRow = MapInputRowParser.parse(inputRowSchema, theMap);
+ }
+
+ @Override
+ public List<String> getDimensions()
+ {
+ return delegateRow.getDimensions();
+ }
+
+ @Override
+ public long getTimestampFromEpoch()
+ {
+ return delegateRow.getTimestampFromEpoch();
+ }
+
+ @Override
+ public DateTime getTimestamp()
+ {
+ return delegateRow.getTimestamp();
+ }
+
+ @Override
+ public List<String> getDimension(String dimension)
+ {
+ return delegateRow.getDimension(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Object getRaw(String dimension)
+ {
+ return delegateRow.getRaw(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Number getMetric(String metric)
+ {
+ return delegateRow.getMetric(metric);
+ }
+
+ @Override
+ public int compareTo(Row o)
+ {
+ return this.getTimestamp().compareTo(o.getTimestamp());
+ }
+
+ @Override
+ public String toString()
+ {
+ return "DeltaInputRow{" +
+ "row=" + row +
+ ", schema=" + schema +
+ ", fieldNameToOrdinal=" + fieldNameToOrdinal +
+ ", delegateRow=" + delegateRow +
+ '}';
+ }
+
+ public Map<String, Object> getRawRowAsMap()
+ {
+ return RowSerde.convertRowToJsonObject(row);
+ }
+
+ @Nullable
+ private Object _getRaw(String dimension)
+ {
+ StructField field = schema.get(dimension);
+ if (field == null || field.isMetadataColumn()) {
+ return null;
+ }
+
+ int ordinal = fieldNameToOrdinal.getInt(dimension);
+ if (ordinal < 0) {
+ return null;
+ }
+ return getValue(field.getDataType(), row, ordinal);
+ }
+
+ @Nullable
+ private static Object getValue(DataType dataType, io.delta.kernel.data.Row dataRow, int columnOrdinal)
+ {
+ if (dataRow.isNullAt(columnOrdinal)) {
Review Comment:
## Chain of 'instanceof' tests
This if block performs a chain of 12 type tests - consider alternatives, e.g. polymorphism or the visitor pattern.
[Show more details](https://github.com/apache/druid/security/code-scanning/6509)
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
Review Comment:
## Inconsistent compareTo
This class declares [compareTo](1) but inherits equals; the two could be inconsistent.
[Show more details](https://github.com/apache/druid/security/code-scanning/6510)
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/RowSerde.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.druid.delta.input;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.delta.kernel.client.TableClient;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.internal.data.DefaultJsonRow;
+import io.delta.kernel.internal.types.TableSchemaSerDe;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.MapType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import org.apache.druid.error.InvalidInput;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility class to serialize and deserialize {@link Row} object.
+ * Code borrowed from <a href="https://github.com/delta-io/delta/blob/master/kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/utils/RowSerDe.java">
+ * RowSerde.java</a>.
+ *
+ */
+public class RowSerde
+{
+ private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();
+
+ private RowSerde()
+ {
+ }
+
+ /**
+ * Utility method to serialize a {@link Row} as a JSON string
+ */
+ public static String serializeRowToJson(Row row)
+ {
+ Map<String, Object> rowObject = convertRowToJsonObject(row);
+ try {
+ Map<String, Object> rowWithSchema = new HashMap<>();
+ rowWithSchema.put("schema", TableSchemaSerDe.toJson(row.getSchema()));
+ rowWithSchema.put("row", rowObject);
+ return OBJECT_MAPPER.writeValueAsString(rowWithSchema);
+ }
+ catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ /**
+ * Utility method to deserialize a {@link Row} object from the JSON form.
+ */
+ public static Row deserializeRowFromJson(TableClient tableClient, String jsonRowWithSchema)
+ {
+ try {
+ JsonNode jsonNode = OBJECT_MAPPER.readTree(jsonRowWithSchema);
+ JsonNode schemaNode = jsonNode.get("schema");
+ StructType schema =
+ TableSchemaSerDe.fromJson(tableClient.getJsonHandler(), schemaNode.asText());
+ return parseRowFromJsonWithSchema((ObjectNode) jsonNode.get("row"), schema);
+ }
+ catch (JsonProcessingException e) {
+ throw new RuntimeException(e);
+ }
+ }
+
+ public static Map<String, Object> convertRowToJsonObject(Row row)
+ {
+ StructType rowType = row.getSchema();
+ Map<String, Object> rowObject = new HashMap<>();
+ for (int fieldId = 0; fieldId < rowType.length(); fieldId++) {
+ StructField field = rowType.at(fieldId);
+ DataType fieldType = field.getDataType();
+ String name = field.getName();
+
+ if (row.isNullAt(fieldId)) {
+ rowObject.put(name, null);
+ continue;
+ }
+
+ Object value;
+ if (fieldType instanceof BooleanType) {
Review Comment:
## Chain of 'instanceof' tests
This if block performs a chain of 13 type tests - consider alternatives, e.g. polymorphism or the visitor pattern.
[Show more details](https://github.com/apache/druid/security/code-scanning/6508)
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on code in PR #15755:
URL: https://github.com/apache/druid/pull/15755#discussion_r1466201507
##########
docs/ingestion/input-sources.md:
##########
@@ -715,6 +715,13 @@ rolled-up datasource `wikipedia_rollup` by grouping on hour, "countryName", and
to `true` to enable a compatibility mode where the timestampSpec is ignored.
:::
+The [secondary partitioning method](native-batch.md#partitionsspec) determines the requisite number of concurrent worker tasks that run in parallel to complete ingestion with the Combining input source.
+Set this value in `maxNumConcurrentSubTasks` in `tuningConfig` based on the secondary partitioning method:
+- `range` or `single_dim` partitioning: greater than or equal to 1
+- `hashed` or `dynamic` partitioning: greater than or equal to 2
+
+For more information on the `maxNumConcurrentSubTasks` field, see [Implementation considerations](native-batch.md#implementation-considerations).
Review Comment:
seems unrelated to the PR
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputSource.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.druid.delta.input;
+
+import com.fasterxml.jackson.annotation.JsonCreator;
+import com.fasterxml.jackson.annotation.JsonProperty;
+import com.google.common.collect.Iterators;
+import com.google.common.primitives.Ints;
+import io.delta.kernel.Scan;
+import io.delta.kernel.Snapshot;
+import io.delta.kernel.Table;
+import io.delta.kernel.TableNotFoundException;
+import io.delta.kernel.client.TableClient;
+import io.delta.kernel.data.FilteredColumnarBatch;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.client.DefaultTableClient;
+import io.delta.kernel.internal.util.Utils;
+import io.delta.kernel.utils.CloseableIterator;
+import org.apache.druid.data.input.InputFormat;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.InputSource;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.data.input.SplitHintSpec;
+import org.apache.druid.data.input.impl.SplittableInputSource;
+import org.apache.druid.error.InvalidInput;
+import org.apache.druid.utils.Streams;
+import org.apache.hadoop.conf.Configuration;
+
+import javax.annotation.Nullable;
+import java.io.File;
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * Input source to ingest data from a Delta Lake.
+ * This input source reads the latest snapshot from a Delta table specified by {@code tablePath} parameter.
+ * Note: the kernel table API only supports reading from the latest snapshot.
+ */
+public class DeltaInputSource implements SplittableInputSource<DeltaSplit>
+{
+ public static final String TYPE_KEY = "delta";
+
+ @JsonProperty
+ private final String tablePath;
+
+ @JsonProperty
+ @Nullable
+ private final DeltaSplit deltaSplit;
+
+ @JsonCreator
+ public DeltaInputSource(
+ @JsonProperty("tablePath") String tablePath,
+ @JsonProperty("deltaSplit") @Nullable DeltaSplit deltaSplit
+ )
+ {
+ if (tablePath == null) {
+ throw InvalidInput.exception("tablePath cannot be null");
+ }
+ this.tablePath = tablePath;
+ this.deltaSplit = deltaSplit;
+ }
+
+ @Override
+ public boolean needsFormat()
+ {
+ // Only support Parquet
+ return false;
+ }
+
+ @Override
+ public InputSourceReader reader(
+ InputRowSchema inputRowSchema,
+ @Nullable InputFormat inputFormat,
+ File temporaryDirectory
+ )
+ {
+ Configuration hadoopConf = new Configuration();
+ TableClient tableClient = DefaultTableClient.create(hadoopConf);
+ try {
+ final Row scanState;
+ final List<Row> scanRowList;
+
Review Comment:
can you add some comments describing the implementation?
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
+{
+ private final io.delta.kernel.data.Row row;
+ private final StructType schema;
+ private final Object2IntMap<String> fieldNameToOrdinal = new Object2IntOpenHashMap<>();
+ private final InputRow delegateRow;
+
+ private static final ZoneId ZONE_ID = ZoneId.systemDefault();
+
+ public DeltaInputRow(io.delta.kernel.data.Row row, InputRowSchema inputRowSchema)
+ {
+ this.row = row;
+ this.schema = row.getSchema();
+ List<String> fieldNames = this.schema.fieldNames();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ fieldNameToOrdinal.put(fieldNames.get(i), i);
+ }
+ fieldNameToOrdinal.defaultReturnValue(-1);
+
+ Map<String, Object> theMap = new HashMap<>();
+ for (String fieldName : fieldNames) {
+ theMap.put(fieldName, _getRaw(fieldName));
+ }
+ delegateRow = MapInputRowParser.parse(inputRowSchema, theMap);
+ }
+
+ @Override
+ public List<String> getDimensions()
+ {
+ return delegateRow.getDimensions();
+ }
+
+ @Override
+ public long getTimestampFromEpoch()
+ {
+ return delegateRow.getTimestampFromEpoch();
+ }
+
+ @Override
+ public DateTime getTimestamp()
+ {
+ return delegateRow.getTimestamp();
+ }
+
+ @Override
+ public List<String> getDimension(String dimension)
+ {
+ return delegateRow.getDimension(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Object getRaw(String dimension)
+ {
+ return delegateRow.getRaw(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Number getMetric(String metric)
+ {
+ return delegateRow.getMetric(metric);
+ }
+
+ @Override
+ public int compareTo(Row o)
+ {
+ return this.getTimestamp().compareTo(o.getTimestamp());
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ return o instanceof DeltaInputRow && compareTo((DeltaInputRow) o) == 0;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(row, schema, fieldNameToOrdinal, delegateRow);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "DeltaInputRow{" +
+ "row=" + row +
+ ", schema=" + schema +
+ ", fieldNameToOrdinal=" + fieldNameToOrdinal +
+ ", delegateRow=" + delegateRow +
+ '}';
+ }
+
+ public Map<String, Object> getRawRowAsMap()
+ {
+ return RowSerde.convertRowToJsonObject(row);
+ }
+
+ @Nullable
+ private Object _getRaw(String dimension)
+ {
+ StructField field = schema.get(dimension);
+ if (field == null || field.isMetadataColumn()) {
+ return null;
+ }
+
+ int ordinal = fieldNameToOrdinal.getInt(dimension);
+ if (ordinal < 0) {
+ return null;
+ }
+ return getValue(field.getDataType(), row, ordinal);
+ }
+
+ @Nullable
+ private static Object getValue(DataType dataType, io.delta.kernel.data.Row dataRow, int columnOrdinal)
+ {
+ if (dataRow.isNullAt(columnOrdinal)) {
+ return null;
+ } else if (dataType instanceof BooleanType) {
+ return dataRow.getBoolean(columnOrdinal);
+ } else if (dataType instanceof ByteType) {
+ return dataRow.getByte(columnOrdinal);
+ } else if (dataType instanceof ShortType) {
+ return dataRow.getShort(columnOrdinal);
+ } else if (dataType instanceof IntegerType) {
+ return dataRow.getInt(columnOrdinal);
+ } else if (dataType instanceof DateType) {
+ // DateType data is stored internally as the number of days since 1970-01-01
+ int daysSinceEpochUTC = dataRow.getInt(columnOrdinal);
+ return LocalDate.ofEpochDay(daysSinceEpochUTC).atStartOfDay(ZONE_ID).toEpochSecond();
+ } else if (dataType instanceof LongType) {
+ return dataRow.getLong(columnOrdinal);
+ } else if (dataType instanceof TimestampType) {
+ // TimestampType data is stored internally as the number of microseconds since epoch
+ long microSecsSinceEpochUTC = dataRow.getLong(columnOrdinal);
+ LocalDateTime dateTime = LocalDateTime.ofEpochSecond(
+ microSecsSinceEpochUTC / 1_000_000 /* epochSecond */,
+ (int) (1000 * microSecsSinceEpochUTC % 1_000_000) /* nanoOfSecond */,
+ ZoneOffset.UTC
+ );
+ return dateTime.atZone(ZONE_ID).toInstant().toEpochMilli();
+ } else if (dataType instanceof FloatType) {
+ return dataRow.getFloat(columnOrdinal);
+ } else if (dataType instanceof DoubleType) {
+ return dataRow.getDouble(columnOrdinal);
+ } else if (dataType instanceof StringType) {
+ return dataRow.getString(columnOrdinal);
+ } else if (dataType instanceof BinaryType) {
+ final byte[] arr = dataRow.getBinary(columnOrdinal);
+ final char[] charArray = new char[arr.length];
+ for (int i = 0; i < arr.length; i++) {
+ charArray[i] = (char) (arr[i] & 0xff);
+ }
+ return String.valueOf(charArray);
+ } else if (dataType instanceof DecimalType) {
+ return dataRow.getDecimal(columnOrdinal).longValue();
+ } else {
+ throw InvalidInput.exception("Unsupported data type[%s]", dataType);
Review Comment:
we should fieldName to the exception message.
##########
extensions-contrib/druid-deltalake-extensions/src/test/java/org/apache/druid/delta/input/DeltaInputSourceTest.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.druid.delta.input;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class DeltaInputSourceTest
+{
+ @Test
+ public void testReadDeltaLakeFilesSample() throws IOException
+ {
+ final DeltaInputSource deltaInputSource = new DeltaInputSource(DeltaTestUtil.DELTA_TABLE_PATH, null);
+ Assert.assertNotNull(deltaInputSource);
+
+ InputSourceReader inputSourceReader = deltaInputSource.reader(DeltaTestUtil.SCHEMA, null, null);
+ Assert.assertNotNull(inputSourceReader);
Review Comment:
inputSourceReader can never be null.
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/DeltaInputRow.java:
##########
@@ -0,0 +1,219 @@
+/*
+ * 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.druid.delta.input;
+
+import io.delta.kernel.types.BinaryType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DecimalType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import it.unimi.dsi.fastutil.objects.Object2IntMap;
+import it.unimi.dsi.fastutil.objects.Object2IntOpenHashMap;
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowSchema;
+import org.apache.druid.data.input.Row;
+import org.apache.druid.data.input.impl.MapInputRowParser;
+import org.apache.druid.error.InvalidInput;
+import org.joda.time.DateTime;
+
+import javax.annotation.Nullable;
+import java.time.LocalDate;
+import java.time.LocalDateTime;
+import java.time.ZoneId;
+import java.time.ZoneOffset;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Encodes the row and schema information from the Delta Lake.
+ */
+public class DeltaInputRow implements InputRow
+{
+ private final io.delta.kernel.data.Row row;
+ private final StructType schema;
+ private final Object2IntMap<String> fieldNameToOrdinal = new Object2IntOpenHashMap<>();
+ private final InputRow delegateRow;
+
+ private static final ZoneId ZONE_ID = ZoneId.systemDefault();
+
+ public DeltaInputRow(io.delta.kernel.data.Row row, InputRowSchema inputRowSchema)
+ {
+ this.row = row;
+ this.schema = row.getSchema();
+ List<String> fieldNames = this.schema.fieldNames();
+ for (int i = 0; i < fieldNames.size(); ++i) {
+ fieldNameToOrdinal.put(fieldNames.get(i), i);
+ }
+ fieldNameToOrdinal.defaultReturnValue(-1);
+
+ Map<String, Object> theMap = new HashMap<>();
+ for (String fieldName : fieldNames) {
+ theMap.put(fieldName, _getRaw(fieldName));
+ }
+ delegateRow = MapInputRowParser.parse(inputRowSchema, theMap);
+ }
+
+ @Override
+ public List<String> getDimensions()
+ {
+ return delegateRow.getDimensions();
+ }
+
+ @Override
+ public long getTimestampFromEpoch()
+ {
+ return delegateRow.getTimestampFromEpoch();
+ }
+
+ @Override
+ public DateTime getTimestamp()
+ {
+ return delegateRow.getTimestamp();
+ }
+
+ @Override
+ public List<String> getDimension(String dimension)
+ {
+ return delegateRow.getDimension(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Object getRaw(String dimension)
+ {
+ return delegateRow.getRaw(dimension);
+ }
+
+ @Nullable
+ @Override
+ public Number getMetric(String metric)
+ {
+ return delegateRow.getMetric(metric);
+ }
+
+ @Override
+ public int compareTo(Row o)
+ {
+ return this.getTimestamp().compareTo(o.getTimestamp());
+ }
+
+ @Override
+ public boolean equals(Object o)
+ {
+ return o instanceof DeltaInputRow && compareTo((DeltaInputRow) o) == 0;
+ }
+
+ @Override
+ public int hashCode()
+ {
+ return Objects.hash(row, schema, fieldNameToOrdinal, delegateRow);
+ }
+
+ @Override
+ public String toString()
+ {
+ return "DeltaInputRow{" +
+ "row=" + row +
+ ", schema=" + schema +
+ ", fieldNameToOrdinal=" + fieldNameToOrdinal +
+ ", delegateRow=" + delegateRow +
+ '}';
+ }
+
+ public Map<String, Object> getRawRowAsMap()
+ {
+ return RowSerde.convertRowToJsonObject(row);
+ }
+
+ @Nullable
+ private Object _getRaw(String dimension)
+ {
+ StructField field = schema.get(dimension);
+ if (field == null || field.isMetadataColumn()) {
+ return null;
+ }
+
+ int ordinal = fieldNameToOrdinal.getInt(dimension);
+ if (ordinal < 0) {
+ return null;
+ }
+ return getValue(field.getDataType(), row, ordinal);
+ }
+
+ @Nullable
+ private static Object getValue(DataType dataType, io.delta.kernel.data.Row dataRow, int columnOrdinal)
+ {
+ if (dataRow.isNullAt(columnOrdinal)) {
+ return null;
+ } else if (dataType instanceof BooleanType) {
+ return dataRow.getBoolean(columnOrdinal);
+ } else if (dataType instanceof ByteType) {
+ return dataRow.getByte(columnOrdinal);
+ } else if (dataType instanceof ShortType) {
+ return dataRow.getShort(columnOrdinal);
+ } else if (dataType instanceof IntegerType) {
+ return dataRow.getInt(columnOrdinal);
+ } else if (dataType instanceof DateType) {
+ // DateType data is stored internally as the number of days since 1970-01-01
+ int daysSinceEpochUTC = dataRow.getInt(columnOrdinal);
+ return LocalDate.ofEpochDay(daysSinceEpochUTC).atStartOfDay(ZONE_ID).toEpochSecond();
+ } else if (dataType instanceof LongType) {
+ return dataRow.getLong(columnOrdinal);
+ } else if (dataType instanceof TimestampType) {
+ // TimestampType data is stored internally as the number of microseconds since epoch
+ long microSecsSinceEpochUTC = dataRow.getLong(columnOrdinal);
+ LocalDateTime dateTime = LocalDateTime.ofEpochSecond(
+ microSecsSinceEpochUTC / 1_000_000 /* epochSecond */,
+ (int) (1000 * microSecsSinceEpochUTC % 1_000_000) /* nanoOfSecond */,
+ ZoneOffset.UTC
+ );
+ return dateTime.atZone(ZONE_ID).toInstant().toEpochMilli();
+ } else if (dataType instanceof FloatType) {
+ return dataRow.getFloat(columnOrdinal);
+ } else if (dataType instanceof DoubleType) {
+ return dataRow.getDouble(columnOrdinal);
+ } else if (dataType instanceof StringType) {
+ return dataRow.getString(columnOrdinal);
+ } else if (dataType instanceof BinaryType) {
+ final byte[] arr = dataRow.getBinary(columnOrdinal);
+ final char[] charArray = new char[arr.length];
+ for (int i = 0; i < arr.length; i++) {
+ charArray[i] = (char) (arr[i] & 0xff);
+ }
+ return String.valueOf(charArray);
Review Comment:
can you describe the logic? why does a single byte need to be masked with `0xff`?
##########
extensions-contrib/druid-deltalake-extensions/src/test/java/org/apache/druid/delta/input/DeltaInputSourceTest.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.druid.delta.input;
+
+import org.apache.druid.data.input.InputRow;
+import org.apache.druid.data.input.InputRowListPlusRawValues;
+import org.apache.druid.data.input.InputSourceReader;
+import org.apache.druid.data.input.InputSplit;
+import org.apache.druid.java.util.common.parsers.CloseableIterator;
+import org.junit.Assert;
+import org.junit.Test;
+
+import java.io.IOException;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+public class DeltaInputSourceTest
+{
+ @Test
+ public void testReadDeltaLakeFilesSample() throws IOException
+ {
+ final DeltaInputSource deltaInputSource = new DeltaInputSource(DeltaTestUtil.DELTA_TABLE_PATH, null);
+ Assert.assertNotNull(deltaInputSource);
Review Comment:
why? :)
##########
extensions-contrib/druid-deltalake-extensions/src/test/resources/create_delta_table.py:
##########
@@ -0,0 +1,89 @@
+#!/usr/bin/env python3
+
+# 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.
+
+import os
+
+import argparse
+import delta
+import mimesis
+import pyspark
+
+
+def config_spark_with_delta_lake():
Review Comment:
if this file is borrowed from some place, please acknowledge the source in comments.
##########
extensions-contrib/druid-deltalake-extensions/src/main/java/org/apache/druid/delta/input/RowSerde.java:
##########
@@ -0,0 +1,155 @@
+/*
+ * 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.druid.delta.input;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.databind.node.ObjectNode;
+import io.delta.kernel.client.TableClient;
+import io.delta.kernel.data.Row;
+import io.delta.kernel.defaults.internal.data.DefaultJsonRow;
+import io.delta.kernel.internal.types.TableSchemaSerDe;
+import io.delta.kernel.internal.util.VectorUtils;
+import io.delta.kernel.types.ArrayType;
+import io.delta.kernel.types.BooleanType;
+import io.delta.kernel.types.ByteType;
+import io.delta.kernel.types.DataType;
+import io.delta.kernel.types.DateType;
+import io.delta.kernel.types.DoubleType;
+import io.delta.kernel.types.FloatType;
+import io.delta.kernel.types.IntegerType;
+import io.delta.kernel.types.LongType;
+import io.delta.kernel.types.MapType;
+import io.delta.kernel.types.ShortType;
+import io.delta.kernel.types.StringType;
+import io.delta.kernel.types.StructField;
+import io.delta.kernel.types.StructType;
+import io.delta.kernel.types.TimestampType;
+import org.apache.druid.error.InvalidInput;
+
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * Utility class to serialize and deserialize {@link Row} object.
+ * Code borrowed from <a href="https://github.com/delta-io/delta/blob/master/kernel/examples/table-reader/src/main/java/io/delta/kernel/examples/utils/RowSerDe.java">
Review Comment:
can you add whats different here from the original class?
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org
Re: [PR] Extension to read and ingest Delta Lake tables (druid)
Posted by "abhishekagarwal87 (via GitHub)" <gi...@apache.org>.
abhishekagarwal87 commented on PR #15755:
URL: https://github.com/apache/druid/pull/15755#issuecomment-1908618232
Is Kerberos required for the first cut?
--
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@druid.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@druid.apache.org
For additional commands, e-mail: commits-help@druid.apache.org