Extension to read and ingest Delta Lake tables

[abhishekrb19]

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.

Example of a DML query using MSQ:

REPLACE INTO "delta-employee-datasource" OVERWRITE ALL
SELECT * FROM TABLE(
  EXTERN(
    '{"type":"delta","tablePath":"s3a://your-bucket/employee-delta-table"}',
    '{"type":"json"}'
  )
) EXTEND ("timestamp" VARCHAR, "name" VARCHAR, "age" BIGINT, "salary" DOUBLE, "bonus" FLOAT)
PARTITIONED BY ALL

Delta ioConfig in native batch spec :

    "ioConfig": {
      "type": "index_parallel",
      "inputSource": {
        "type": "delta",
        "tablePath": "s3a://your-bucket/employee-delta-table"
      },
    }

Web-console screenshots:

Sample and ingestion of a mock dataset from the cloud:

Delta 3.1.0 just released. Can look into upgrading the delta dependency as a follow-up.

Release note

Enhanced ingestion capabilities to support ingestion of Delta Lake tables.

This PR has:

• been self-reviewed.
  • using the concurrency checklist (Remove this item if the PR doesn't have any relation to concurrency.)
• added documentation for new or modified features or behaviors.
• a release note entry in the PR description.
• added Javadocs for most classes and all non-trivial methods. Linked related entities via Javadoc links.
• added or updated version, license, or notice information in licenses.yaml
• added comments explaining the "why" and the intent of the code wherever would not be obvious for an unfamiliar reader.
• added unit tests or modified existing tests to cover new code paths, ensuring the threshold for code coverage is met.
• added integration tests.
• been tested in a test Druid cluster.

[vogievetsky]

Web console changes look good.

[vtlim]

Some nits but docs otherwise look good!

[pjain1]

@abhishekrb19 thanks for this. I was just going through this and wondering if delta lake kernel api supports any partitioning by time that we can leverage during indexing. Ingestion spec generally have an interval so can that be used to prune the amount of files on delta lake that is read during indexing, with my limited understanding of delta lake kernel api it supports filtering on columns while building scan.

I have not seen many delta lake data lakes but one I seen had a base directory which hosted the metadata in _delta_log folder, data was partitioned by day so had subdirectories for each day of data and those directories has hourly subdirectories. So if someone needs to run hourly indexing job will the extension end up reading all data again and again ?

[abhishekrb19]

Hi @pjain1, yes, Delta tables can be partitioned by date. Coincidentally, Delta 3.1.0 released today. In 3.1.0, the kernel supports data skipping in addition to partition pruning using filter predicates, whenever applicable. I will try to find time and add this functionality soon. In its current form, the connector will do full table scans, so I think upgrading to 3.1.0 and adding support for filters will be a good next step.

[pjain1]

@abhishekrb19 thanks