Package ai.rapids.cudf

Class HybridScanReader

java.lang.Object
ai.rapids.cudf.HybridScanReader
All Implemented Interfaces:
AutoCloseable
@Experimental public class HybridScanReader extends Object implements AutoCloseable
Experimental Parquet hybrid-scan reader.

This class is the Java binding for cudf::io::parquet::experimental::hybrid_scan_reader. It is designed for highly selective filter expressions over Parquet files and exposes the multi-step pipeline that the C++ reader uses internally:

  1. Read the Parquet footer (and optional page index) to drive row-group / page-level pruning.
  2. Filter the row groups using statistics, bloom filters, and dictionary pages.
  3. Materialize the filter columns; the reader builds an initial row mask (all true or seeded from page index stats when page-level pruning is enabled) and then mutates it down to only the rows that survive the AST filter.
  4. Materialize the payload columns using the surviving row mask.

A two-step convenience flow is provided for selective filters (materializeFilterColumns(int[], DeviceMemoryBuffer[], boolean) then materializePayloadColumns(int[], DeviceMemoryBuffer[], ColumnVector, boolean)), and a one-shot materializeAllColumns(int[], DeviceMemoryBuffer[]) is provided for small files or broad filters.

Chunked / streaming materialization is also supported via the setupChunkingFor* + materialize*Chunk family of methods, mirroring the C++ chunked reader pipeline.

The filter and payload materialization paths accept a boolean that toggles page-level pruning: skips decode of pages the filter (or row mask) proves empty, in exchange for a per-page stats scan and a carried row-mask column. Enable when the workload prunes many pages; on the filter path this requires prior setupPageIndex(HostMemoryBuffer).

The reader is created with no filter expression installed. Filter-related APIs behave as though nothing has been filtered out unless a filter is first supplied via setFilter(CompiledExpression).

The APIs in this file are experimental and subject to change.

  • Constructor Details

  • Method Details

    • setFilter

      public void setFilter(CompiledExpression filter)
      Install or replace the filter expression used by all subsequent filter-related APIs on this reader; pass null to clear. Cheap to call, so callers can sweep the same file with several candidate predicates in a loop.

      The caller retains ownership of any previously installed CompiledExpression and must close it themselves — this method replaces this reader's strong reference only.

      Parameters:
      filter - the new filter expression, or null to clear

    • pageIndexByteRange

      public ByteRange pageIndexByteRange()
      Returns:
      the byte range of the page index in the Parquet file.

    • setupPageIndex

      public void setupPageIndex(HostMemoryBuffer pageIndexBuffer)
      Materialize the ColumnIndex / OffsetIndex structs (collectively, the page index) from the supplied bytes. Required before any filter or payload materialization call with usePageLevelPruning == true.
      Parameters:
      pageIndexBuffer - host-resident page index bytes

    • allRowGroups

      public int[] allRowGroups()
      Returns:
      all row group indices in the Parquet file.

    • totalRowsInRowGroups

      public long totalRowsInRowGroups(int[] rowGroupIndices)
      Returns:
      the total number of top-level rows in the supplied row groups.

    • filterRowGroupsWithStats

      public int[] filterRowGroupsWithStats(int[] rowGroupIndices)
      Filter row groups using column-chunk statistics from the file footer.

    • secondaryFiltersByteRanges

      public SecondaryFilterRanges secondaryFiltersByteRanges(int[] rowGroupIndices)
      Get the byte ranges in the source file that hold the bloom filter and dictionary page data needed for the next round of row-group pruning.

    • filterRowGroupsWithDictionaryPages

      public int[] filterRowGroupsWithDictionaryPages(DeviceMemoryBuffer[] dictionaryPageData, int[] rowGroupIndices)
      Filter row groups using column-chunk dictionary pages loaded into device memory.

    • filterColumnChunksByteRanges

      public ByteRange[] filterColumnChunksByteRanges(int[] rowGroupIndices)
      Returns:
      byte ranges for the column chunks of filter columns.

    • payloadColumnChunksByteRanges

      public ByteRange[] payloadColumnChunksByteRanges(int[] rowGroupIndices)
      Returns:
      byte ranges for the column chunks of payload columns.

      This result is order-dependent. If filter columns have already been processed on this reader (e.g. via filterColumnChunksByteRanges(int[]) or materializeFilterColumns(int[], DeviceMemoryBuffer[], boolean)), the filter columns are excluded and only the payload columns are returned. If filter columns have not yet been processed, nothing is excluded and the ranges cover the full set of columns that would be read i.e. the columns projected via ParquetOptions, or all columns in the file when no projection was set.

    • allColumnChunksByteRanges

      public ByteRange[] allColumnChunksByteRanges(int[] rowGroupIndices)
      Returns:
      byte ranges for all column chunks (filter + payload) of the selected columns.

    • materializeFilterColumns

      public HybridScanReader.FilterMaterializationResult materializeFilterColumns(int[] rowGroupIndices, DeviceMemoryBuffer[] columnChunkData, boolean usePageLevelPruning)
      Build the initial row mask, materialize the filter columns, and evaluate the compiled filter expression against them. The row mask and the resulting filter table are returned together as a HybridScanReader.FilterMaterializationResult; close it via try-with-resources.

      Set usePageLevelPruning = true to skip decompression and decode of pages the filter proves empty. Requires prior setupPageIndex(HostMemoryBuffer).

      Cost: a per-page stats scan of filter columns and a carried row-mask column. Payoff: pruned pages are skipped entirely, typically the dominant read cost. Enable when a meaningful fraction of pages can be pruned; otherwise leave false.

      Parameters:
      rowGroupIndices - row groups to read
      columnChunkData - device buffers holding the filter column chunks, in the order returned by filterColumnChunksByteRanges(int[])
      usePageLevelPruning - seed the row mask from page-index stats and enable the data page mask; requires prior setupPageIndex(HostMemoryBuffer)
      Returns:
      combined filter table and mutated row mask; caller must close this result

    • materializePayloadColumns

      public Table materializePayloadColumns(int[] rowGroupIndices, DeviceMemoryBuffer[] columnChunkData, ColumnVector rowMask, boolean usePageLevelPruning)
      Materialize only the payload columns, applying the supplied row mask to the output.

      Set usePageLevelPruning = true to skip decompression and decode of pages containing no rows surviving rowMask. Adds a small mask-build cost; wins when rowMask prunes a meaningful fraction of pages.

      Parameters:
      rowGroupIndices - row groups to read
      columnChunkData - device buffers holding the payload column chunks, in the order returned by payloadColumnChunksByteRanges(int[])
      rowMask - row mask (read-only)
      usePageLevelPruning - enable the data page mask to skip decode of pages the row mask proves empty; requires prior setupPageIndex(HostMemoryBuffer)
      Returns:
      the materialized payload column table

    • materializeAllColumns

      public Table materializeAllColumns(int[] rowGroupIndices, DeviceMemoryBuffer[] columnChunkData)
      Materialize all (or selected) columns in a single pass. The filter expression is applied after reading, so this is most efficient for small files or when most rows survive the filter. For highly-selective filters, prefer the explicit two-step flow.

    • setupChunkingForFilterColumns

      public void setupChunkingForFilterColumns(long chunkReadLimit, long passReadLimit, int[] rowGroupIndices, boolean usePageLevelPruning, DeviceMemoryBuffer[] columnChunkData)
      Build the initial row mask and set up chunking state for filter-column materialization. The row mask is owned internally by the reader for the duration of the chunked filter pipeline; subsequent calls to materializeFilterColumnsChunk() mutate it in place. After all chunks are drained, call takeFilterRowMask() to transfer ownership of the row mask to the caller (typically to feed it into materializePayloadColumns(int[], DeviceMemoryBuffer[], ColumnVector, boolean)).

      Set usePageLevelPruning = true to skip decompression and decode of pages the filter proves empty. Requires prior setupPageIndex(HostMemoryBuffer).

      Cost: a per-page stats scan of filter columns and a carried row-mask column. Payoff: pruned pages are skipped entirely, typically the dominant read cost. Enable when a meaningful fraction of pages can be pruned; otherwise leave false.

      Calling this method again before takeFilterRowMask() discards the previous chunked-filter row mask.

      Caller must keep columnChunkData open until takeFilterRowMask() (or a re-setup); the native reader holds references to it across chunk calls.

      Parameters:
      chunkReadLimit - soft limit (in bytes) on each output chunk returned by the matching materialize*Chunk call, or 0 for no limit
      passReadLimit - soft limit (in bytes) on the working memory used to decompress and decode a single subpass (a page-level slice of the selected row groups), or 0 for no limit. Hybrid scan hardcodes the number of row-group passes to 1, so this parameter only bounds the subpass (decode) working set; it does not repartition row groups. To split row groups across multiple passes up front, call constructRowGroupPasses(int[], long) first and issue a separate chunked run per returned partition.
      rowGroupIndices - row groups to read
      usePageLevelPruning - seed the row mask from page-index stats and enable the data page mask; requires prior setupPageIndex(HostMemoryBuffer)
      columnChunkData - device buffers holding the filter column chunks, in the order returned by filterColumnChunksByteRanges(int[])

    • materializeFilterColumnsChunk

      public Table materializeFilterColumnsChunk()
      Returns:
      the next filter-column chunk; throws if no chunk is available or if no chunked filter pipeline is active. The internal row mask owned by this reader is updated in place.

    • takeFilterRowMask

      public ColumnVector takeFilterRowMask()
      Transfer ownership of the row mask produced by the chunked filter-column materialization to the caller. Typically called after all materializeFilterColumnsChunk() calls complete and immediately before materializePayloadColumns(int[], DeviceMemoryBuffer[], ColumnVector, boolean).

      After this call the reader has no chunked-filter row mask; subsequent materializeFilterColumnsChunk() calls will fail until setupChunkingForFilterColumns(long, long, int[], boolean, DeviceMemoryBuffer[]) is invoked again. If never called, the row mask is freed when the reader is closed.

      Returns:
      the owned row mask ColumnVector; caller must close it.

    • setupChunkingForPayloadColumns

      public void setupChunkingForPayloadColumns(long chunkReadLimit, long passReadLimit, int[] rowGroupIndices, ColumnVector rowMask, boolean usePageLevelPruning, DeviceMemoryBuffer[] columnChunkData)
      Set up chunking state for payload-column materialization. Subsequent calls to materializePayloadColumnsChunk(ColumnVector) will yield successive chunks.

      Set usePageLevelPruning = true to skip decompression and decode of pages containing no rows surviving rowMask. Adds a small mask-build cost; wins when rowMask prunes a meaningful fraction of pages.

      Caller must keep columnChunkData open until hasNextTableChunk() returns false; the native reader holds references to it across chunk calls.

      Parameters:
      chunkReadLimit - soft limit (in bytes) on each output chunk returned by the matching materialize*Chunk call, or 0 for no limit
      passReadLimit - soft limit (in bytes) on the working memory used to decompress and decode a single subpass (a page-level slice of the selected row groups), or 0 for no limit. Hybrid scan hardcodes the number of row-group passes to 1, so this parameter only bounds the subpass (decode) working set; it does not repartition row groups. To split row groups across multiple passes up front, call constructRowGroupPasses(int[], long) first and issue a separate chunked run per returned partition.
      rowGroupIndices - row groups to read
      rowMask - row mask (read-only)
      usePageLevelPruning - enable the data page mask to skip decode of pages the row mask proves empty; requires prior setupPageIndex(HostMemoryBuffer)
      columnChunkData - device buffers holding the payload column chunks, in the order returned by payloadColumnChunksByteRanges(int[])

    • materializePayloadColumnsChunk

      public Table materializePayloadColumnsChunk(ColumnVector rowMask)
      Returns:
      the next payload-column chunk; throws if no chunk is available.

    • setupChunkingForAllColumns

      public void setupChunkingForAllColumns(long chunkReadLimit, long passReadLimit, int[] rowGroupIndices, DeviceMemoryBuffer[] columnChunkData)
      Set up chunking state for all-column materialization (single-pass mode). Subsequent calls to materializeAllColumnsChunk() will yield successive chunks.

      Caller must keep columnChunkData open until hasNextTableChunk() returns false; the native reader holds references to it across chunk calls.

      Parameters:
      chunkReadLimit - soft limit (in bytes) on each output chunk returned by the matching materialize*Chunk call, or 0 for no limit
      passReadLimit - soft limit (in bytes) on the working memory used to decompress and decode a single subpass (a page-level slice of the selected row groups), or 0 for no limit. Hybrid scan hardcodes the number of row-group passes to 1, so this parameter only bounds the subpass (decode) working set; it does not repartition row groups. To split row groups across multiple passes up front, call constructRowGroupPasses(int[], long) first and issue a separate chunked run per returned partition.
      rowGroupIndices - row groups to read
      columnChunkData - device buffers holding all column chunks, in the order returned by allColumnChunksByteRanges(int[])

    • materializeAllColumnsChunk

      public Table materializeAllColumnsChunk()
      Returns:
      the next all-columns chunk; throws if no chunk is available.

    • hasNextTableChunk

      public boolean hasNextTableChunk()
      Returns:
      true when a subsequent materialize*Chunk call would return data.

    • constructRowGroupPasses

      public int[][] constructRowGroupPasses(int[] rowGroupIndices, long passReadLimit)
      Partition the supplied row groups into passes whose total uncompressed size respects the given limit. The returned array contains one inner array per pass.
      Parameters:
      rowGroupIndices - row groups to partition
      passReadLimit - limit on the memory used by a single pass, or 0 for no limit. Each returned pass can then be fed to a setupChunkingFor* call, which will further stream that pass in subpass-sized chunks bounded by its own passReadLimit argument.
      Returns:
      an array of arrays of row group indices, one per pass

    • close

      public void close()
      Specified by:
      close in interface AutoCloseable