Repository: DeepRec-AI/HybridBackend (language: python)

Issue:
ParquetDataset support configuration with default value
# User Story

The fixed-length features in TFRecord support configuration with default values（https://www.tensorflow.org/api_docs/python/tf/io/FixedLenFeature）, but currently, Parquet does not support this feature. If encountering a non-existent feature, an error will be thrown, such as 'No Column called xx found in yy'. 

So when we transition from training with TFRecord to Parquet, there may be some functionality alignment issues. We hope that Parquet can also support configuration with default values.



Discussion / hints:

Fix patch:
diff --git a/docs/data.md b/docs/data.md
index dce996fd..79c5ded4 100644
--- a/docs/data.md
+++ b/docs/data.md
@@ -152,6 +152,7 @@ import hybridbackend.tensorflow as hb
 # Define data frame fields.
 fields = [
     hb.data.DataFrame.Field('A', tf.int64),  # scalar
+    hb.data.DataFrame.Field('AA', tf.int64, default_value=42),  # field not exists
     hb.data.DataFrame.Field('B', tf.int64, shape=[32]),  # fixed-length list
     hb.data.DataFrame.Field('C', tf.int64, ragged_rank=1),  # variable-length list
     hb.data.DataFrame.Field('D', tf.int64, ragged_rank=1)]  # variable-length list
@@ -164,7 +165,7 @@ ds = ds.batch(1024)
 ds = ds.prefetch(4)
 it = tf.data.make_one_shot_iterator(ds)
 batch = it.get_next()
-# {'A': scalar_tensor, 'B': list_tensor, 'C': sparse_tensor, 'D': padded_list_tensor}
+# {'A': scalar_tensor, 'AA': [42] * 1024, 'B': list_tensor, 'C': sparse_tensor, 'D': padded_list_tensor}
 ...
 ```
 
diff --git a/hybridbackend/tensorflow/data/__init__.py b/hybridbackend/tensorflow/data/__init__.py
index ad49f79d..74610a26 100644
--- a/hybridbackend/tensorflow/data/__init__.py
+++ b/hybridbackend/tensorflow/data/__init__.py
@@ -23,6 +23,7 @@
 
 from hybridbackend.tensorflow.data.dataframe import DataFrame
 from hybridbackend.tensorflow.data.dataframe import parse
+from hybridbackend.tensorflow.data.dataframe import populate_defaults
 from hybridbackend.tensorflow.data.dataframe import unbatch_and_to_sparse
 from hybridbackend.tensorflow.data.deduplicate.dataset import deduplicate
 from hybridbackend.tensorflow.data.prefetch.iterator import Iterator
diff --git a/hybridbackend/tensorflow/data/dataframe.py b/hybridbackend/tensorflow/data/dataframe.py
index 2943eb36..e30aca60 100644
--- a/hybridbackend/tensorflow/data/dataframe.py
+++ b/hybridbackend/tensorflow/data/dataframe.py
@@ -155,7 +155,8 @@ def _to_legacy_output_classes(self):
           [ops.Tensor for i in xrange(self._field.ragged_rank)])
 
     def __init__(
-        self, name, dtype=None, ragged_rank=None, shape=None):
+        self, name, dtype=None, ragged_rank=None, shape=None,
+        default_value=None):
       self._name = name
       if dtype is None:
         self._dtype = dtype
@@ -181,6 +182,7 @@ def __init__(
       else:
         shape = tensor_shape.TensorShape([])
       self._shape = shape
+      self._default_value = default_value
       self._restore_idx_field = None
 
     @property
@@ -203,6 +205,10 @@ def ragged_rank(self):
     def shape(self):
       return self._shape
 
+    @property
+    def default_value(self):
+      return self._default_value
+
     @property
     def restore_idx_field(self):
       return self._restore_idx_field
@@ -223,7 +229,12 @@ def __repr__(self):
           elif self._ragged_rank > 0:
             dtypestr = f'list<{dtypestr}>'
       shapestr = str(self._shape)
-      return f'{self._name} (dtype={dtypestr}, shape={shapestr})'
+      defaultstr = str(self._default_value)
+      return (
+        f'{self._name} ('
+        f'dtype={dtypestr}, '
+        f'shape={shapestr}, '
+        f'default_value={defaultstr})')
 
     def map(self, func, rank=None):
       if rank is None:
@@ -444,6 +455,51 @@ def parse(cls, features, pad=False, restore_idx=None):
       return features
     raise ValueError(f'{features} not supported')
 
+  @classmethod
+  def populate_defaults(cls, features, all_fields, batch_size):
+    r'''Populate default values.
+    '''
+    if not isinstance(features, dict):
+      raise ValueError('Inputs should be a dict')
+    populated = dict(features)
+    for f in all_fields:
+      if f.name not in features and f.default_value is not None:
+        if batch_size is None:
+          populated[f.name] = array_ops.expand_dims(
+            ops.convert_to_tensor(f.default_value, dtype=f.dtype),
+            axis=0)
+        else:
+          if isinstance(f.default_value, sparse_tensor.SparseTensor):
+            indices_dtype = f.default_value.indices.dtype
+            indices_car = math_ops.cast(
+              array_ops.reshape(
+                array_ops.tile(
+                  math_ops.range(batch_size),
+                  [f.default_value.indices.shape[0]]),
+                [-1, 1]),
+              indices_dtype)
+            indices_cdr = array_ops.tile(
+              f.default_value.indices, [batch_size, 1])
+            batched_indices = array_ops.concat(
+              [indices_car, indices_cdr], axis=1)
+            batched_values = array_ops.tile(
+              f.default_value.values, [batch_size])
+            batched_dense_shape = array_ops.concat(
+              [ops.convert_to_tensor([batch_size], indices_dtype),
+               f.default_value.dense_shape],
+              axis=0)
+            populated[f.name] = sparse_tensor.SparseTensor(
+              indices=batched_indices,
+              values=batched_values,
+              dense_shape=batched_dense_shape)
+          else:
+            value = ops.convert_to_tensor(f.default_value, dtype=f.dtype)
+            populated[f.name] = array_ops.tile(
+              array_ops.expand_dims(value, axis=0),
+              tensor_shape.TensorShape([batch_size]).concatenate(
+                value.shape))
+    return populated
+
   @classmethod
   def to_sparse(cls, features):
     r'''Convert DataFrame values to tensors or sparse tensors.
@@ -509,6 +565,16 @@ def _apply_fn(dataset):
   return _apply_fn
 
 
+def populate_defaults(all_fields, batch_size, num_parallel_calls=None):
+  r'''Populate default values.
+  '''
+  def _apply_fn(dataset):
+    return dataset.map(
+      lambda t: DataFrame.populate_defaults(t, all_fields, batch_size),
+      num_parallel_calls=num_parallel_calls)
+  return _apply_fn
+
+
 def unbatch_and_to_sparse(num_parallel_calls=None):
   r'''Unbatch and convert a row to tensors or sparse tensors from input dataset.
   '''
@@ -554,9 +620,6 @@ def build_fields(filename, fn, fields=None, lower=False):
   '''
   logging.info(f'Reading fields from {filename} ...')
   all_field_tuples = fn(filename)  # pylint: disable=c-extension-no-member
-  if not all_field_tuples:
-    raise ValueError(
-      f'No supported fields found in file {filename}')
   all_fields = {
     f[0]: {'dtype': f[1], 'ragged_rank': f[2]}
     for f in all_field_tuples}
@@ -573,28 +636,66 @@ def build_fields(filename, fn, fields=None, lower=False):
           shape=f.shape,
           ragged_rank=f.ragged_rank)
       if f.name not in all_fields:
-        raise ValueError(
-          f'Field {f.name} is not found in the file {filename}')
+        if f.default_value is None:
+          raise ValueError(
+            f'Field {f.name} not found in file {filename}')
       dtype = f.dtype
-      actual_dtype = np.dtype(all_fields[f.name]['dtype'])
-      if dtype is None:
-        dtype = actual_dtype
-      elif dtype != actual_dtype:
-        raise ValueError(
-          f'Field {f.name} should has dtype {actual_dtype} not {dtype}')
-      ragged_rank = f.ragged_rank
-      actual_ragged_rank = all_fields[f.name]['ragged_rank']
-      if ragged_rank is None:
-        ragged_rank = actual_ragged_rank
-      elif ragged_rank != actual_ragged_rank:
-        raise ValueError(
-          f'Field {f.name} should has ragged_rank {actual_ragged_rank} '
-          f'not {ragged_rank}')
+
+      if f.name in all_fields:
+        actual_dtype = np.dtype(all_fields[f.name]['dtype'])
+        if dtype is None:
+          dtype = actual_dtype
+        elif dtype != actual_dtype:
+          raise ValueError(
+            f'Field {f.name} dtype should be {actual_dtype} not {dtype}')
+        ragged_rank = f.ragged_rank
+        actual_ragged_rank = all_fields[f.name]['ragged_rank']
+        if ragged_rank is None:
+          ragged_rank = actual_ragged_rank
+        elif ragged_rank != actual_ragged_rank:
+          raise ValueError(
+            f'Field {f.name} ragged_rank should be {actual_ragged_rank} '
+            f'not {ragged_rank}')
+      else:
+        if f.default_value is None:
+          raise ValueError(
+            f'Field {f.name} not found in file {filename}')
+        if isinstance(f.default_value, sparse_tensor.SparseTensor):
+          actual_dtype = np.dtype(f.default_value.dtype)
+          if dtype is None:
+            dtype = actual_dtype
+          elif dtype != actual_dtype:
+            raise ValueError(
+              f'Field {f.name} dtype should be {actual_dtype} not {dtype}')
+        elif isinstance(f.default_value, ops.Tensor):
+          actual_dtype = np.dtype(f.default_value.dtype)
+          if dtype is None:
+            dtype = actual_dtype
+          elif dtype != actual_dtype:
+            raise ValueError(
+              f'Field {f.name} dtype should be {actual_dtype} not {dtype}')
+          actual_ragged_rank = 0
+          if ragged_rank is None:
+            ragged_rank = actual_ragged_rank
+          elif ragged_rank != actual_ragged_rank:
+            raise ValueError(
+              f'Field {f.name} ragged_rank should be {actual_ragged_rank} '
+              f'not {ragged_rank}')
+        else:
+          try:
+            with ops.name_scope('default_values/'):
+              _ = ops.convert_to_tensor(
+                f.default_value, dtype=dtype)
+          except (TypeError, ValueError) as ex:
+            raise ValueError(
+              f'Field {f.name} default_value {f.default_value} '
+              f'should be a SparseTensor or Tensor: {ex}') from ex
       f = DataFrame.Field(
         f.name,
         dtype=dtype,
         ragged_rank=ragged_rank,
-        shape=f.shape)
+        shape=f.shape,
+        default_value=None if f.name in all_fields else f.default_value)
       new_fields.append(f)
       continue
     if not isinstance(f, string):
diff --git a/hybridbackend/tensorflow/data/tabular/dataset_v1.py b/hybridbackend/tensorflow/data/tabular/dataset_v1.py
index bf7acb40..8f4483a0 100644
--- a/hybridbackend/tensorflow/data/tabular/dataset_v1.py
+++ b/hybridbackend/tensorflow/data/tabular/dataset_v1.py
@@ -197,6 +197,10 @@ def output_classes(self):
   def creator(self):
     return self._creator
 
+  @property
+  def fields(self):
+    return self.creator.fields
+
   def batch(self, batch_size, drop_remainder=False):
     r'''Combines consecutive elements of this dataset into batches.
 
@@ -256,7 +260,7 @@ def shuffle_batch(
       reshuffle_each_iteration=reshuffle_each_iteration)
 
   @abc.abstractmethod
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
 
@@ -302,7 +306,6 @@ def __init__(
     '''
     filenames, fields = build_filenames_and_fields(
       filenames, parquet_file_get_fields, fields, lower=field_ignore_case)
-    self._fields = fields
     super().__init__(
       filenames,
       fields=fields,
@@ -316,16 +319,12 @@ def __init__(
       field_map_fn=field_map_fn,
       **kwargs)
 
-  @property
-  def fields(self):
-    return self._fields
-
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
     return _TabularDatasetV1(  # pylint: disable=abstract-class-instantiated
       TableFormats.PARQUET, filename, batch_size,
-      fields=self.creator.fields,
+      fields=fields,
       partition_count=self.creator.partition_count,
       partition_index=self.creator.partition_index,
       skip_corrupted_data=self.creator.skip_corrupted_data,
@@ -373,7 +372,6 @@ def __init__(
     '''
     filenames, fields = build_filenames_and_fields(
       filenames, orc_file_get_fields, fields, lower=field_ignore_case)
-    self._fields = fields
     super().__init__(
       filenames,
       fields=fields,
@@ -387,16 +385,12 @@ def __init__(
       field_map_fn=field_map_fn,
       **kwargs)
 
-  @property
-  def fields(self):
-    return self._fields
-
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
     return _TabularDatasetV1(  # pylint: disable=abstract-class-instantiated
       TableFormats.ORC, filename, batch_size,
-      fields=self.creator.fields,
+      fields=fields,
       partition_count=self.creator.partition_count,
       partition_index=self.creator.partition_index,
       skip_corrupted_data=self.creator.skip_corrupted_data,
@@ -458,8 +452,9 @@ def __init__(
         can only be set once. if `tf.data.experimental.AUTOTUNE` is used, the
         number of parsers would be set for best performance.
     '''
-    filenames, self._fields = build_filenames_and_fields(
+    filenames, self._all_fields = build_filenames_and_fields(
       filenames, parquet_file_get_fields, fields)
+    self._fields = [f for f in self._all_fields if f.default_value is None]
     self._partition_count = partition_count
     self._partition_index = partition_index
     self._skip_corrupted_data = skip_corrupted_data
@@ -499,6 +494,10 @@ def _create_dataset(f):
       _create_dataset, filenames, num_parallel_reads, num_sequential_reads)
     super().__init__()
 
+  @property
+  def all_fields(self):
+    return self._all_fields
+
   @property
   def fields(self):
     return self._fields
diff --git a/hybridbackend/tensorflow/data/tabular/dataset_v2.py b/hybridbackend/tensorflow/data/tabular/dataset_v2.py
index 21257787..054eacc6 100644
--- a/hybridbackend/tensorflow/data/tabular/dataset_v2.py
+++ b/hybridbackend/tensorflow/data/tabular/dataset_v2.py
@@ -177,6 +177,10 @@ def element_spec(self):
   def creator(self):
     return self._creator
 
+  @property
+  def fields(self):
+    return self.creator.fields
+
   def batch(self, batch_size, drop_remainder=False):
     r'''Combines consecutive elements of this dataset into batches.
 
@@ -236,7 +240,7 @@ def shuffle_batch(
       reshuffle_each_iteration=reshuffle_each_iteration)
 
   @abc.abstractmethod
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
 
@@ -282,7 +286,6 @@ def __init__(
     '''
     filenames, fields = build_filenames_and_fields(
       filenames, parquet_file_get_fields, fields, lower=field_ignore_case)
-    self._fields = fields
     super().__init__(
       filenames,
       fields=fields,
@@ -296,16 +299,12 @@ def __init__(
       field_map_fn=field_map_fn,
       **kwargs)
 
-  @property
-  def fields(self):
-    return self._fields
-
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
     return _TabularDatasetV2(  # pylint: disable=abstract-class-instantiated
       TableFormats.PARQUET, filename, batch_size,
-      fields=self.creator.fields,
+      fields=fields,
       partition_count=self.creator.partition_count,
       partition_index=self.creator.partition_index,
       skip_corrupted_data=self.creator.skip_corrupted_data,
@@ -353,7 +352,6 @@ def __init__(
     '''
     filenames, fields = build_filenames_and_fields(
       filenames, orc_file_get_fields, fields, lower=field_ignore_case)
-    self._fields = fields
     super().__init__(
       filenames,
       fields=fields,
@@ -367,16 +365,12 @@ def __init__(
       field_map_fn=field_map_fn,
       **kwargs)
 
-  @property
-  def fields(self):
-    return self._fields
-
-  def _create_dataset(self, filename, batch_size):
+  def _create_dataset(self, filename, fields, batch_size):
     r'''Internal method to create a `TabularDataset`.
     '''
     return _TabularDatasetV2(  # pylint: disable=abstract-class-instantiated
       TableFormats.ORC, filename, batch_size,
-      fields=self.creator.fields,
+      fields=fields,
       partition_count=self.creator.partition_count,
       partition_index=self.creator.partition_index,
       skip_corrupted_data=self.creator.skip_corrupted_data,
@@ -438,8 +432,9 @@ def __init__(
         can only be set once. if `tf.data.experimental.AUTOTUNE` is used, the
         number of parsers would be set for best performance.
     '''
-    filenames, self._fields = build_filenames_and_fields(
+    filenames, self._all_fields = build_filenames_and_fields(
       filenames, parquet_file_get_fields, fields)
+    self._fields = [f for f in self._all_fields if f.default_value is None]
     self._partition_count = partition_count
     self._partition_index = partition_index
     self._skip_corrupted_data = skip_corrupted_data
@@ -481,6 +476,10 @@ def _create_dataset(f):
       num_sequential_reads=num_sequential_reads)
     super().__init__(self._impl._variant_tensor)  # pylint: disable=protected-access
 
+  @property
+  def all_fields(self):
+    return self._all_fields
+
   @property
   def fields(self):
     return self._fields
diff --git a/hybridbackend/tensorflow/data/tabular/table.py b/hybridbackend/tensorflow/data/tabular/table.py
index 8f74973a..dc1fc340 100644
--- a/hybridbackend/tensorflow/data/tabular/table.py
+++ b/hybridbackend/tensorflow/data/tabular/table.py
@@ -29,6 +29,7 @@
 from tensorflow.python.framework import ops
 
 from hybridbackend.tensorflow.data.dataframe import parse
+from hybridbackend.tensorflow.data.dataframe import populate_defaults
 from hybridbackend.tensorflow.data.deduplicate.dataset import deduplicate
 from hybridbackend.tensorflow.data.rebatch.dataset import RebatchDataset
 
@@ -80,6 +81,7 @@ def __init__(
     self._fn = fn
     self._filenames = filenames
     self._fields = fields if field_map_fn is None else field_map_fn(fields)
+    self._valid_fields = [f for f in self._fields if f.default_value is None]
     self._partition_count = partition_count
     self._partition_index = partition_index
     self._skip_corrupted_data = skip_corrupted_data
@@ -95,9 +97,9 @@ def __init__(
       else:
         num_parallel_reads = 1
     if num_parallel_parser_calls is None:
-      num_parallel_parser_calls = len(self._fields)
+      num_parallel_parser_calls = len(self._valid_fields)
     if num_parallel_parser_calls == dataset_ops.AUTOTUNE:
-      num_parallel_parser_calls = len(self._fields)
+      num_parallel_parser_calls = len(self._valid_fields)
     if num_parallel_reads is not None and num_parallel_parser_calls is not None:
       arrow_num_threads = os.getenv('ARROW_NUM_THREADS', None)
       if arrow_num_threads is None:
@@ -159,7 +161,7 @@ def _create_dataset(self, batch_size):
     '''
     def _creator(filename):
       filename = ops.convert_to_tensor(filename, dtypes.string, name='filename')
-      return self._fn(filename, batch_size)
+      return self._fn(filename, self._valid_fields, batch_size)
 
     if self._num_parallel_reads == 1:
       return self._filenames.flat_map(_creator)
@@ -184,8 +186,9 @@ def read(self):
       ds = ds.apply(
         deduplicate(
           self._key_idx_field_names,
-          self._value_field_names, fields=self._fields))
+          self._value_field_names, fields=self._valid_fields))
     ds = ds.apply(parse(pad=self._to_dense))
+    ds = ds.apply(populate_defaults(self._fields, None))
     return ds.unbatch()
 
   def batch(self, batch_size, drop_remainder=False):
@@ -217,11 +220,12 @@ def batch(self, batch_size, drop_remainder=False):
       ds = ds.apply(
         deduplicate(
           self._key_idx_field_names,
-          self._value_field_names, fields=self._fields))
+          self._value_field_names, fields=self._valid_fields))
     ds = RebatchDataset(
-      ds, self._fields, batch_size,
+      ds, self._valid_fields, batch_size,
       drop_remainder=drop_remainder)
-    return ds.apply(parse(pad=self._to_dense))
+    ds = ds.apply(parse(pad=self._to_dense))
+    return ds.apply(populate_defaults(self._fields, batch_size))
 
   def shuffle_batch(
       self, batch_size,
@@ -260,11 +264,12 @@ def shuffle_batch(
       ds = ds.apply(
         deduplicate(
           self._key_idx_field_names,
-          self._value_field_names, fields=self._fields))
+          self._value_field_names, fields=self._valid_fields))
     ds = RebatchDataset(
-      ds, self._fields, batch_size,
+      ds, self._valid_fields, batch_size,
       drop_remainder=drop_remainder,
       shuffle_buffer_size=buffer_size,
       shuffle_seed=seed,
       reshuffle_each_iteration=reshuffle_each_iteration)
-    return ds.apply(parse(pad=self._to_dense))
+    ds = ds.apply(parse(pad=self._to_dense))
+    return ds.apply(populate_defaults(self._fields, batch_size))


Test patch:
diff --git a/hybridbackend/tensorflow/data/tests/parquet_dataset_deduplicate_test.py b/hybridbackend/tensorflow/data/tests/parquet_dataset_deduplicate_test.py
index 58022e83..a5de4cba 100644
--- a/hybridbackend/tensorflow/data/tests/parquet_dataset_deduplicate_test.py
+++ b/hybridbackend/tensorflow/data/tests/parquet_dataset_deduplicate_test.py
@@ -76,11 +76,12 @@ def test_apply_to_sparse(self):
 
   def test_apply_to_tensor(self):
     with tf.Graph().as_default() as graph:
-      ds = hb.data.ParquetDataset(
+      srcds = hb.data.ParquetDataset(
         [self._filename],
         batch_size=2)
-      ds = ds.apply(hb.data.deduplicate(['user_feat_idx'], [['user_feat']]))
+      ds = srcds.apply(hb.data.deduplicate(['user_feat_idx'], [['user_feat']]))
       ds = ds.apply(hb.data.parse(pad=True))
+      ds = ds.apply(hb.data.populate_defaults(srcds.all_fields, 2))
       batch = tf.data.make_one_shot_iterator(ds).get_next()
       baseline = tf.ragged.constant(
         self._data_user_feat_duplicated.to_pylist()).to_tensor()
diff --git a/hybridbackend/tensorflow/data/tests/parquet_dataset_test.py b/hybridbackend/tensorflow/data/tests/parquet_dataset_test.py
index 00e2f27a..b4ce6855 100644
--- a/hybridbackend/tensorflow/data/tests/parquet_dataset_test.py
+++ b/hybridbackend/tensorflow/data/tests/parquet_dataset_test.py
@@ -74,6 +74,32 @@ def test_read(self):
         np.testing.assert_equal(result['A'], a[start_row:end_row].to_numpy())
         np.testing.assert_equal(result['C'], c[start_row:end_row].to_numpy())
 
+  def test_read_with_defaults(self):
+    batch_size = 32
+    default_value = 42
+    with tf.Graph().as_default() as graph:
+      ds = hb.data.Dataset.from_parquet(
+        self._filename,
+        fields=[hb.data.DataFrame.Field('A', tf.int64),
+                hb.data.DataFrame.Field('X', tf.int64,
+                                        default_value=default_value),
+                hb.data.DataFrame.Field('C', tf.int64)])
+      ds = ds.batch(batch_size)
+      ds = ds.prefetch(4)
+      batch = tf.data.make_one_shot_iterator(ds).get_next()
+
+    a = self._df['A']
+    c = self._df['C']
+    with tf.Session(graph=graph) as sess:
+      for i in xrange(3):
+        result = sess.run(batch)
+        start_row = i * batch_size
+        end_row = (i + 1) * batch_size
+        np.testing.assert_equal(result['A'], a[start_row:end_row].to_numpy())
+        np.testing.assert_equal(
+          result['X'], [default_value for _ in range(batch_size)])
+        np.testing.assert_equal(result['C'], c[start_row:end_row].to_numpy())
+
   def test_schema_auto_detection_read(self):
     batch_size = 32
     with tf.Graph().as_default() as graph: