Python connector CDF reads kernel integration #613
Conversation
PatrickJin-db
force-pushed
the
PatrickJin-db/cdf_python_changes
branch
7 times, most recently
from
eda025a to
75835d3
Compare
| @@ -4,17 +4,19 @@ use arrow::compute::filter_record_batch; | |||
| use arrow::datatypes::SchemaRef; | |||
There was a problem hiding this comment.
Note: Both arrow and kernel have their own SchemaRef. This is up to personal preference, but you may want to give this an alias so that the type of SchemaRef doesn't become confusing.
| use arrow::datatypes::SchemaRef; | |
| use arrow::datatypes::SchemaRef as ArrowSchemaRef; |
There was a problem hiding this comment.
You may also want to do the same for TableChangesScan:
use delta_kernel::table_changes::scan::TableChangesScan as KernelTableChangesScanThere was a problem hiding this comment.
I think ArrowSchemaRef makes sense, but TableChangesScan is only used once so I don't think using an alias for it is as beneficial.
| ).table_changes_to_pandas(CdfOptions( | ||
| starting_version=starting_version, | ||
| ending_version=ending_version, | ||
| starting_timestamp=starting_timestamp, | ||
| ending_timestamp=ending_timestamp, | ||
| include_historical_metadata=use_delta_format |
There was a problem hiding this comment.
hmmm? why include_historical_metadata=use_delta_format?
There was a problem hiding this comment.
I see why, let's add a comment to this line about it.
| # os.utime accepts seconds while delta log timestamp is in ms | ||
| os.utime(log_file_path, times=(0, version_to_timestamp[version] // 1000)) | ||
|
|
||
| if min_version > 0 and num_versions_with_action > 0: |
There was a problem hiding this comment.
In what case will num_versions_with_action be 0?
There was a problem hiding this comment.
One case could be if only metadata changes occurred during the queried version range.
| protocol_json = json.loads(next(lines)) | ||
| metadata_json = json.loads(next(lines)) | ||
| actions: List[FileAction] = [] | ||
| for line in lines: |
There was a problem hiding this comment.
and let's try some logs for the ease of debugging in the future when issues happen.
Such as printing out # of lines received from the server.
PatrickJin-db
force-pushed
the
PatrickJin-db/cdf_python_changes
branch
8 times, most recently
from
6cb527d to
e72d39f
Compare
| ).table_changes_to_pandas(CdfOptions( | ||
| starting_version=starting_version, | ||
| ending_version=ending_version, | ||
| starting_timestamp=starting_timestamp, | ||
| ending_timestamp=ending_timestamp, | ||
| include_historical_metadata=use_delta_format |
PatrickJin-db
force-pushed
the
PatrickJin-db/cdf_python_changes
branch
8 times, most recently
from
22b77cc to
4032dc6
Compare
PatrickJin-db
force-pushed
the
PatrickJin-db/cdf_python_changes
branch
from
4032dc6 to
4a9f4db
Compare
Main differences compared to the prototype #609 include:
Regarding performance, most of the time is spent in kernel, and writing the delta log is negligible in comparison.
