More updates for VCF ingestion with tiledb URIs (#624)

src/tiledb/cloud/utilities/consolidate.py

@@ -101,6 +101,7 @@ def consolidate_and_vacuum(
  logger = get_logger()
 
  with tiledb.scope_ctx(config):
+ # TODO: remove when remote fragment consolidation is supported
  is_remote = array_uri.startswith("tiledb://")
  if not is_remote and vacuum_fragments:
  logger.info("Vacuuming fragments")

src/tiledb/cloud/vcf/ingestion.py

@@ -900,6 +900,10 @@ def consolidate_dataset_udf(
  group = tiledb.Group(dataset_uri)
 
  for member in group:
+ # Skip non-array members
+ if member.type != tiledb.Array:
+ continue
+
  uri = member.uri
  name = member.name
  is_remote = uri.startswith("tiledb://")
@@ -916,13 +920,15 @@ def consolidate_dataset_udf(
  modes += ["fragments"]
 
  for mode in modes:
+ logger.debug("Consolidating %r in %r (%s)", mode, uri, name)
  config = tiledb.Config({"sm.consolidation.mode": mode})
  try:
  tiledb.consolidate(uri, config=config)
  except Exception as e:
  print(e)
 
  for mode in modes:
+ logger.debug("Vacuuming %r in %r (%s)", mode, uri, name)
  config = tiledb.Config({"sm.vacuum.mode": mode})
  try:
  tiledb.vacuum(uri, config=config)
@@ -1315,7 +1321,9 @@ def ingest_samples_dag(
  consolidate.depends_on(ingest)
 
  # Consolidate fragments in the stats arrays, if enabled
- if consolidate_stats:
+ # TODO: remove when remote fragment consolidation is supported
+ is_remote = dataset_uri.startswith("tiledb://")
+ if consolidate_stats and not is_remote:
 
  def group_member_uri(group_uri, group_member, config):
  with tiledb.scope_ctx(config):