feat: Alternative table provider as spark data source

[timsaucer]

DO NOT MERGE

This PR is so drafty you can't stand next to it without wearing a sweater.

Alternative to #103 attempting to follow the suggestion in: #104 (comment)

[paleolimbot]

Thank you for putting this together! It's a good thing I live in an old house in Winnipeg and own a number of excellent sweaters.

I left a few suggestions but in general I like this approach, or perhaps, I feel as though I can do a better job reviewing this than a more JNI heavy approach because I know the Arrow C interfaces better than I know JNI. In this approach we are basically building a mini ADBC but it's a constrained subset of the things ADBC is trying to do, so I think it is OK.

When there's consensus on the approach I'm happy to review any of them 🙂

[paleolimbot]

Mostly just highlighting for future me the key pieces here...this is the JNI piece. I'm biased because I don't know JNI but I do know Arrow, so the fact that this version only has a few hundred lines of JNI appeals to me.

[paleolimbot]

I think you can do this without the C symbols (just structs, like the Arrow C Data/Stream interfaces). The reason that's nice is that symbols can be hard to guarantee uniqueness (what if you have two Rust crates that want to export table provider and they're statically linked in the same pile of crates?). This is roughly how datafusion-python works with the generated FFI from datafusion-ffi.

struct DfScanHandle {
    int32_t (*partition_count)(const DfScanHandle* self);
    // -1 partition ID for coalesced stream
    int32_t (*execute)(const DfScanHandle* self, int32_t partition,
                                  struct ArrowArrayStream* out_stream);
    const char* (*get_last_error)(struct DfScanHandle* self);
    void* private_data;
    void (*release)(struct DfScanHandle* self);
};

struct DfSimpleTableProvider {
  int32_t (*schema)(struct DFTableProvider* self, struct ArrowSchema* out_schema);
  int32_t (*scan_create)(DfStr provider, DfBytes options, DfBytes partition,
                       int32_t target_partitions, int32_t batch_size, int64_t limit,
                       const DfKeyValue* config_overrides, size_t config_overrides_len,
                       const DfStr* projection, size_t projection_len,
                       const DfBytes* filters, size_t filters_len,
                       DfScanHandle** out_handle);
  const char* (*get_last_error)(struct DFTableProvider* self);
  void* private_data;
  void (*release)(struct DFTableProvider* self);
};

For Spark, a TableProvider exporter could define an entrypoint and build a cdylib. The entrypoint would have to have a common signature (like how AdbcInitFunc works).

int32_t MyProviderInit(void* table_provider, const char* name, int32_t df_abi_version);

...but Python packages that export one of these could just work with capsules.

[timsaucer]

In this approach we are basically building a mini ADBC but it's a constrained subset of the things ADBC is trying to do

Yes, but I'm starting to think more about this. I did some analysis with my agent of where the gaps in the ADBC were and it was suggesting the problems with not having explicit push down filters. However I think if we use the substrait interface instead of the SQL interface then that argument disappears. I'm slightly outside my comfort zone on some of these aspects.

What I'm going to try next - instead lean in towards your suggestion of ADBC and see if that can support all of the spark options I'd need and still get the pushdown. Also I need to make sure the task/partition mapping would still work.

[paleolimbot]

You're kind to try all these!

However I think if we use the substrait interface instead of the SQL interface then that argument disappears.

Also note that while these names are "substrait" and "SQL", the only API requirement is that the first one is "bytes" and the second one is "string". We can and should (personal opinion) abuse them (e.g., statement.set_option("datafusion.query_format", "protobuf"), statement.set_option("datafusion.statement.filters", b"xxxxx")).

[tokoko]

hey, just FYI, there's an issue in spark apache/spark#54603 for an adbc data source and a mostly working implementation (with sql pushdown) in a separate repo that you might want to check out.

[timsaucer]

hey, just FYI, there's an issue in spark apache/spark#54603 for an adbc data source and a mostly working implementation (with sql pushdown) in a separate repo that you might want to check out.

Thank you! Looks very interesting. I will definitely follow that PR.