avoid read api where result small

Author: TrevorBergeron

bigframes/core/nodes.py

@@ -743,7 +743,6 @@ def remap_refs(
         return self
 
 
-## Put ordering in here or just add order_by node above?
 @dataclasses.dataclass(frozen=True, eq=False)
 class ReadTableNode(LeafNode):
     source: bq_data.BigqueryDataSource
@@ -753,11 +752,6 @@ class ReadTableNode(LeafNode):
 
     table_session: bigframes.session.Session = dataclasses.field()
 
-    # DELETE BEFORE SUBMITTING, DEBUG ONLY
-    def __post_init__(self):
-        for item in self.scan_list.items:
-            assert item.source_id in self.source.schema.names
-
     def _validate(self):
         # enforce invariants
         physical_names = set(map(lambda i: i.name, self.source.table.physical_schema))

bigframes/session/bq_caching_executor.py

@@ -623,18 +623,14 @@ def _execute_plan_gbq(
 
             create_table = True
             if not cache_spec.cluster_cols:
+                assert len(cache_spec.cluster_cols) <= _MAX_CLUSTER_COLUMNS
                 offsets_id = bigframes.core.identifiers.ColumnId(
                     bigframes.core.guid.generate_guid()
                 )
                 plan = nodes.PromoteOffsetsNode(plan, offsets_id)
                 cluster_cols = [offsets_id.sql]
             else:
-                cluster_cols = [
-                    col
-                    for col in cache_spec.cluster_cols
-                    if bigframes.dtypes.is_clusterable(plan.schema.get_type(col))
-                ]
-                cluster_cols = cluster_cols[:_MAX_CLUSTER_COLUMNS]
+                cluster_cols = cache_spec.cluster_cols
 
         compiled = compile.compile_sql(
             compile.CompileRequest(
@@ -690,7 +686,11 @@ def _execute_plan_gbq(
         execution_metadata = executor.ExecutionMetadata.from_iterator_and_job(
             iterator, query_job
         )
-        if result_bq_data is not None:
+        result_mostly_cached = (
+            hasattr(iterator, "_is_almost_completely_cached")
+            and iterator._is_almost_completely_cached()
+        )
+        if result_bq_data is not None and not result_mostly_cached:
             return executor.BQTableExecuteResult(
                 data=result_bq_data,
                 project_id=self.bqclient.project,