File: C:/Users/fred/anaconda3/Lib/site-packages/tables/tests/test_indexes.py
import copy
import tempfile
from pathlib import Path
import numpy as np
import tables as tb
from tables.tests import common
# Sensible parameters for indexing with small blocksizes
minRowIndex = 10
small_blocksizes = (96, 24, 6, 3)
class TDescr(tb.IsDescription):
var1 = tb.StringCol(itemsize=4, dflt=b"", pos=1)
var2 = tb.BoolCol(dflt=0, pos=2)
var3 = tb.IntCol(dflt=0, pos=3)
var4 = tb.FloatCol(dflt=0, pos=4)
class BasicTestCase(common.TempFileMixin, common.PyTablesTestCase):
compress = 0
complib = "zlib"
shuffle = 0
fletcher32 = 0
nrows = minRowIndex
ss = small_blocksizes[2]
def setUp(self):
super().setUp()
self.rootgroup = self.h5file.root
self.populateFile()
# Close the file
self.h5file.close()
def populateFile(self):
group = self.rootgroup
# Create a table
title = "This is the IndexArray title"
self.filters = tb.Filters(complevel=self.compress,
complib=self.complib,
shuffle=self.shuffle,
fletcher32=self.fletcher32)
table = self.h5file.create_table(group, 'table', TDescr, title,
self.filters, self.nrows)
for i in range(self.nrows):
table.row['var1'] = str(i).encode('ascii')
# table.row['var2'] = i > 2
table.row['var2'] = i % 2
table.row['var3'] = i
table.row['var4'] = float(self.nrows - i - 1)
table.row.append()
table.flush()
# Index all entries:
for col in table.colinstances.values():
indexrows = col.create_index(_blocksizes=small_blocksizes)
if common.verbose:
print("Number of written rows:", self.nrows)
print("Number of indexed rows:", indexrows)
return
def test00_flushLastRow(self):
"""Checking flushing an Index incrementing only the last row."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test00_flushLastRow..." %
self.__class__.__name__)
# Open the HDF5 file in append mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
# Add just 3 rows more
for i in range(3):
table.row['var1'] = str(i).encode('ascii')
table.row.append()
table.flush() # redo the indexes
idxcol = table.cols.var1.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
print("Elements in last row:", idxcol.indicesLR[-1])
# Do a selection
results = [p["var1"] for p in table.where('var1 == b"1"')]
self.assertEqual(len(results), 2)
self.assertEqual(results, [b'1']*2)
def test00_update(self):
"""Checking automatic re-indexing after an update operation."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test00_update..." % self.__class__.__name__)
# Open the HDF5 file in append mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
# Modify a couple of columns
for i, row in enumerate(table.where("(var3>1) & (var3<5)")):
row['var1'] = str(i)
row['var3'] = i
row.update()
table.flush() # redo the indexes
idxcol1 = table.cols.var1.index
idxcol3 = table.cols.var3.index
if common.verbose:
print("Dirtyness of var1 col:", idxcol1.dirty)
print("Dirtyness of var3 col:", idxcol3.dirty)
self.assertEqual(idxcol1.dirty, False)
self.assertEqual(idxcol3.dirty, False)
# Do a couple of selections
results = [p["var1"] for p in table.where('var1 == b"1"')]
self.assertEqual(len(results), 2)
self.assertEqual(results, [b'1']*2)
results = [p["var3"] for p in table.where('var3 == 0')]
self.assertEqual(len(results), 2)
self.assertEqual(results, [0]*2)
def test01_readIndex(self):
"""Checking reading an Index (string flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_readIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-only mode
self.h5file = tb.open_file(self.h5fname, mode="r")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
results = [p["var1"] for p in table.where('var1 == b"1"')]
self.assertEqual(len(results), 1)
self.assertEqual(results, [b'1'])
def test02_readIndex(self):
"""Checking reading an Index (bool flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_readIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-only mode
self.h5file = tb.open_file(self.h5fname, mode="r")
table = self.h5file.root.table
idxcol = table.cols.var2.index
if common.verbose:
print("Rows in table:", table.nrows)
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
results = [p["var2"] for p in table.where('var2 == True')]
if common.verbose:
print("Selected values:", results)
self.assertEqual(len(results), self.nrows // 2)
self.assertEqual(results, [True]*(self.nrows // 2))
def test03_readIndex(self):
"""Checking reading an Index (int flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_readIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-only mode
self.h5file = tb.open_file(self.h5fname, mode="r")
table = self.h5file.root.table
idxcol = table.cols.var3.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
results = [p["var3"] for p in table.where('(1<var3)&(var3<10)')]
if common.verbose:
print("Selected values:", results)
self.assertEqual(len(results), min(10, table.nrows) - 2)
self.assertEqual(results, list(range(2, min(10, table.nrows))))
def test04_readIndex(self):
"""Checking reading an Index (float flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_readIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-only mode
self.h5file = tb.open_file(self.h5fname, mode="r")
table = self.h5file.root.table
idxcol = table.cols.var4.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of rows in table:", table.nrows)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
results = [p["var4"] for p in table.where('var4 < 10')]
# results = [p["var4"] for p in table.where('(1<var4)&(var4<10)')]
if common.verbose:
print("Selected values:", results)
self.assertEqual(len(results), min(10, table.nrows))
self.assertEqual(
results,
[float(i) for i in reversed(list(range(min(10, table.nrows))))])
def test05_getWhereList(self):
"""Checking reading an Index with get_where_list (string flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_getWhereList..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var4.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
table.flavor = "python"
rowList1 = table.get_where_list('var1 < b"10"')
rowList2 = [p.nrow for p in table if p['var1'] < b"10"]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test06_getWhereList(self):
"""Checking reading an Index with get_where_list (bool flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test06_getWhereList..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var2.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Rows in tables:", table.nrows)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
table.flavor = "numpy"
rowList1 = table.get_where_list('var2 == False', sort=True)
rowList2 = [p.nrow for p in table if p['var2'] is False]
# Convert to a NumPy object
rowList2 = np.array(rowList2, np.int64)
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertTrue(common.allequal(rowList1, rowList2))
def test07_getWhereList(self):
"""Checking reading an Index with get_where_list (int flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test07_getWhereList..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var4.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
table.flavor = "python"
rowList1 = table.get_where_list('var3 < 15', sort=True)
rowList2 = [p.nrow for p in table if p["var3"] < 15]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test08_getWhereList(self):
"""Checking reading an Index with get_where_list (float flavor)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08_getWhereList..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var4.index
if common.verbose:
print("Max rows in buf:", table.nrowsinbuf)
print("Number of elements per slice:", idxcol.slicesize)
print("Chunk size:", idxcol.sorted.chunksize)
# Do a selection
table.flavor = "python"
rowList1 = table.get_where_list('var4 < 10', sort=True)
rowList2 = [p.nrow for p in table if p['var4'] < 10]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test09a_removeIndex(self):
"""Checking removing an index."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test09a_removeIndex..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before deletion")
print("var1 column:", table.cols.var1)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# delete the index
table.cols.var1.remove_index()
if common.verbose:
print("After deletion")
print("var1 column:", table.cols.var1)
self.assertIsNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 0)
# re-create the index again
indexrows = table.cols.var1.create_index(_blocksizes=small_blocksizes)
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
if common.verbose:
print("After re-creation")
print("var1 column:", table.cols.var1)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
def test09b_removeIndex(self):
"""Checking removing an index (persistent version)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test09b_removeIndex..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before deletion")
print("var1 index column:", table.cols.var1)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
# delete the index
table.cols.var1.remove_index()
# close and reopen the file
self._reopen(mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("After deletion")
print("var1 column:", table.cols.var1)
self.assertIsNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 0)
# re-create the index again
indexrows = table.cols.var1.create_index(_blocksizes=small_blocksizes)
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
if common.verbose:
print("After re-creation")
print("var1 column:", table.cols.var1)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
def test10a_moveIndex(self):
"""Checking moving a table with an index."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10a_moveIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before move")
print("var1 column:", idxcol)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# Create a new group called "agroup"
agroup = self.h5file.create_group("/", "agroup")
# move the table to "agroup"
table.move(agroup, "table2")
if common.verbose:
print("After move")
print("var1 column:", idxcol)
self.assertIsNotNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 1)
# Some sanity checks
table.flavor = "python"
rowList1 = table.get_where_list('var1 < b"10"')
rowList2 = [p.nrow for p in table if p['var1'] < b"10"]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test10b_moveIndex(self):
"""Checking moving a table with an index (persistent version)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10b_moveIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before move")
print("var1 index column:", idxcol)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
# Create a new group called "agroup"
agroup = self.h5file.create_group("/", "agroup")
# move the table to "agroup"
table.move(agroup, "table2")
# close and reopen the file
self._reopen(mode="a")
table = self.h5file.root.agroup.table2
idxcol = table.cols.var1.index
if common.verbose:
print("After move")
print("var1 column:", idxcol)
self.assertIsNotNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 1)
# Some sanity checks
table.flavor = "python"
rowList1 = table.get_where_list('var1 < b"10"')
rowList2 = [p.nrow for p in table if p['var1'] < b"10"]
if common.verbose:
print("Selected values:", rowList1, type(rowList1))
print("Should look like:", rowList2, type(rowList2))
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test10c_moveIndex(self):
"""Checking moving a table with an index (small node cache)."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10c_moveIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a", node_cache_slots=10)
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before move")
print("var1 column:", idxcol)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# Create a new group called "agroup"
agroup = self.h5file.create_group("/", "agroup")
# move the table to "agroup"
table.move(agroup, "table2")
if common.verbose:
print("After move")
print("var1 column:", idxcol)
self.assertIsNotNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 1)
# Some sanity checks
table.flavor = "python"
rowList1 = table.get_where_list('var1 < b"10"')
rowList2 = [p.nrow for p in table if p['var1'] < b"10"]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test10d_moveIndex(self):
"""Checking moving a table with an index (no node cache)."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10d_moveIndex..." % self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a", node_cache_slots=0)
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before move")
print("var1 column:", idxcol)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# Create a new group called "agroup"
agroup = self.h5file.create_group("/", "agroup")
# move the table to "agroup"
table.move(agroup, "table2")
if common.verbose:
print("After move")
print("var1 column:", idxcol)
self.assertIsNotNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var1"], 1)
# Some sanity checks
table.flavor = "python"
rowList1 = table.get_where_list('var1 < b"10"')
rowList2 = [p.nrow for p in table if p['var1'] < b"10"]
if common.verbose:
print("Selected values:", rowList1)
print("Should look like:", rowList2)
self.assertEqual(len(rowList1), len(rowList2))
self.assertEqual(rowList1, rowList2)
def test11a_removeTableWithIndex(self):
"""Checking removing a table with indexes."""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test11a_removeTableWithIndex..." %
self.__class__.__name__)
# Open the HDF5 file in read-write mode
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before deletion")
print("var1 column:", table.cols.var1)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# delete the table
self.h5file.remove_node("/table")
if common.verbose:
print("After deletion")
self.assertNotIn("table", self.h5file.root)
# re-create the table and the index again
table = self.h5file.create_table("/", 'table', TDescr, "New table",
self.filters, self.nrows)
for i in range(self.nrows):
table.row['var1'] = str(i)
table.row['var2'] = i % 2
table.row['var3'] = i
table.row['var4'] = float(self.nrows - i - 1)
table.row.append()
table.flush()
# Index all entries:
for col in table.colinstances.values():
indexrows = col.create_index(_blocksizes=small_blocksizes)
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
if common.verbose:
print("After re-creation")
print("var1 column:", table.cols.var1)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
def test11b_removeTableWithIndex(self):
"""Checking removing a table with indexes (persistent version 2)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test11b_removeTableWithIndex..." %
self.__class__.__name__)
self.h5file = tb.open_file(self.h5fname, mode="a")
table = self.h5file.root.table
idxcol = table.cols.var1.index
if common.verbose:
print("Before deletion")
print("var1 column:", table.cols.var1)
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
# delete the table
self.h5file.remove_node("/table")
if common.verbose:
print("After deletion")
self.assertNotIn("table", self.h5file.root)
# close and reopen the file
self._reopen(mode="r+")
# re-create the table and the index again
table = self.h5file.create_table("/", 'table', TDescr, "New table",
self.filters, self.nrows)
for i in range(self.nrows):
table.row['var1'] = str(i)
table.row['var2'] = i % 2
table.row['var3'] = i
table.row['var4'] = float(self.nrows - i - 1)
table.row.append()
table.flush()
# Index all entries:
for col in table.colinstances.values():
indexrows = col.create_index(_blocksizes=small_blocksizes)
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
if common.verbose:
print("After re-creation")
print("var1 column:", table.cols.var1)
self.assertIsNotNone(idxcol)
self.assertEqual(table.colindexed["var1"], 1)
# Test provided by Andrew Straw
def test11c_removeTableWithIndex(self):
"""Checking removing a table with indexes (persistent version 3)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test11c_removeTableWithIndex..." %
self.__class__.__name__)
class Distance(tb.IsDescription):
frame = tb.Int32Col(pos=0)
distance = tb.FloatCol(pos=1)
# Delete the old temporal file
Path(self.h5fname).unlink()
self.h5fname = tempfile.mktemp(".h5")
self.h5file = tb.open_file(self.h5fname, mode='w')
table = self.h5file.create_table(
self.h5file.root, 'distance_table', Distance)
table.cols.frame.create_index(_blocksizes=small_blocksizes)
r = table.row
for i in range(10):
r['frame'] = i
r['distance'] = float(i**2)
r.append()
table.flush()
self._reopen(mode='r+')
self.h5file.remove_node(self.h5file.root.distance_table)
def test12_doubleIterate(self):
self.h5file = tb.open_file(self.h5fname, mode="r")
table = self.h5file.root.table
tests = [1, 4, self.nrows]
if self.nrows > 500:
tests.append(self.nrows - 500)
for limit in tests:
handle_a = [0, table.where('(var3 < e)', dict(e=limit))]
handle_b = [0, table.where('(var3 < e)', dict(e=limit))]
try:
while True:
next(handle_b[1])
handle_b[0] += 1
except StopIteration:
for _ in handle_a[1]:
handle_a[0] += 1
for _ in handle_b[1]:
handle_b[0] += 1
self.assertEqual(handle_a[0], limit)
self.assertEqual(handle_b[0], limit)
self.assertEqual(
len(list(table.where('(var3 < e)', dict(e=limit)))), limit)
small_ss = small_blocksizes[2]
class BasicReadTestCase(BasicTestCase):
compress = 0
complib = "zlib"
shuffle = 0
fletcher32 = 0
nrows = small_ss
class ZlibReadTestCase(BasicTestCase):
compress = 1
complib = "zlib"
shuffle = 0
fletcher32 = 0
nrows = small_ss
@common.unittest.skipIf(not common.blosc_avail,
'BLOSC compression library not available')
class BloscReadTestCase(BasicTestCase):
compress = 1
complib = "blosc"
shuffle = 0
fletcher32 = 0
nrows = small_ss
@common.unittest.skipIf(not common.lzo_avail,
'LZO compression library not available')
class LZOReadTestCase(BasicTestCase):
compress = 1
complib = "lzo"
shuffle = 0
fletcher32 = 0
nrows = small_ss
@common.unittest.skipIf(not common.bzip2_avail,
'BZIP2 compression library not available')
class Bzip2ReadTestCase(BasicTestCase):
compress = 1
complib = "bzip2"
shuffle = 0
fletcher32 = 0
nrows = small_ss
class ShuffleReadTestCase(BasicTestCase):
compress = 1
complib = "zlib"
shuffle = 1
fletcher32 = 0
nrows = small_ss
class Fletcher32ReadTestCase(BasicTestCase):
compress = 1
complib = "zlib"
shuffle = 0
fletcher32 = 1
nrows = small_ss
class ShuffleFletcher32ReadTestCase(BasicTestCase):
compress = 1
complib = "zlib"
shuffle = 1
fletcher32 = 1
nrows = small_ss
class OneHalfTestCase(BasicTestCase):
nrows = small_ss + small_ss//2
class UpperBoundTestCase(BasicTestCase):
nrows = small_ss + 1
class LowerBoundTestCase(BasicTestCase):
nrows = small_ss * 2-1
class DeepTableIndexTestCase(common.TempFileMixin, common.PyTablesTestCase):
nrows = minRowIndex
def test01(self):
"""Checking the indexing of a table in a 2nd level hierarchy"""
# Create an instance of an HDF5 Table
group = self.h5file.create_group(self.h5file.root, "agroup")
# Create a table
title = "This is the IndexArray title"
table = self.h5file.create_table(group, 'table', TDescr, title,
None, self.nrows)
for i in range(self.nrows):
# Fill rows with defaults
table.row.append()
table.flush()
# Index some column
indexrows = table.cols.var1.create_index()
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
# Some sanity checks
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
self.assertEqual(idxcol.nelements, self.nrows)
def test01b(self):
"""Checking the indexing of a table in 2nd level
(persistent version)"""
# Create an instance of an HDF5 Table
group = self.h5file.create_group(self.h5file.root, "agroup")
# Create a table
title = "This is the IndexArray title"
table = self.h5file.create_table(group, 'table', TDescr, title,
None, self.nrows)
for i in range(self.nrows):
# Fill rows with defaults
table.row.append()
table.flush()
# Index some column
indexrows = table.cols.var1.create_index()
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
# Close and re-open this file
self._reopen(mode='a')
table = self.h5file.root.agroup.table
idxcol = table.cols.var1.index
# Some sanity checks
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
self.assertEqual(idxcol.nelements, self.nrows)
def test02(self):
"""Checking the indexing of a table in a 4th level hierarchy"""
# Create an instance of an HDF5 Table
group = self.h5file.create_group(self.h5file.root, "agroup")
group = self.h5file.create_group(group, "agroup")
group = self.h5file.create_group(group, "agroup")
# Create a table
title = "This is the IndexArray title"
table = self.h5file.create_table(group, 'table', TDescr, title,
None, self.nrows)
for i in range(self.nrows):
# Fill rows with defaults
table.row.append()
table.flush()
# Index some column
indexrows = table.cols.var1.create_index()
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
# Some sanity checks
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
self.assertEqual(idxcol.nelements, self.nrows)
def test02b(self):
"""Checking the indexing of a table in a 4th level
(persistent version)"""
# Create an instance of an HDF5 Table
group = self.h5file.create_group(self.h5file.root, "agroup")
group = self.h5file.create_group(group, "agroup")
group = self.h5file.create_group(group, "agroup")
# Create a table
title = "This is the IndexArray title"
table = self.h5file.create_table(group, 'table', TDescr, title,
None, self.nrows)
for i in range(self.nrows):
# Fill rows with defaults
table.row.append()
table.flush()
# Index some column
indexrows = table.cols.var1.create_index()
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
# Close and re-open this file
self._reopen(mode='a')
table = self.h5file.root.agroup.agroup.agroup.table
idxcol = table.cols.var1.index
# Some sanity checks
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
self.assertEqual(idxcol.nelements, self.nrows)
def test03(self):
"""Checking the indexing of a table in a 100th level hierarchy"""
# Create an instance of an HDF5 Table
group = self.h5file.root
for i in range(100):
group = self.h5file.create_group(group, "agroup")
# Create a table
title = "This is the IndexArray title"
table = self.h5file.create_table(group, 'table', TDescr, title,
None, self.nrows)
for i in range(self.nrows):
# Fill rows with defaults
table.row.append()
table.flush()
# Index some column
indexrows = table.cols.var1.create_index()
self.assertIsNotNone(indexrows)
idxcol = table.cols.var1.index
# Some sanity checks
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(idxcol)
self.assertEqual(idxcol.nelements, self.nrows)
class IndexProps:
def __init__(self, auto=tb.index.default_auto_index,
filters=tb.index.default_index_filters):
self.auto = auto
self.filters = filters
DefaultProps = IndexProps()
NoAutoProps = IndexProps(auto=False)
ChangeFiltersProps = IndexProps(
filters=tb.Filters(complevel=6, complib="zlib",
shuffle=False, fletcher32=False))
class AutomaticIndexingTestCase(common.TempFileMixin, common.PyTablesTestCase):
reopen = 1
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
small_blocksizes = (16, 8, 4, 2)
def setUp(self):
super().setUp()
# Create an instance of an HDF5 Table
title = "This is the IndexArray title"
root = self.h5file.root
# Make the chunkshape smaller or equal than small_blocksizes[-1]
chunkshape = (2,)
self.table = self.h5file.create_table(root, 'table', TDescr, title,
None, self.nrows,
chunkshape=chunkshape)
self.table.autoindex = self.iprops.auto
for colname in self.colsToIndex:
self.table.colinstances[colname].create_index(
_blocksizes=self.small_blocksizes)
for i in range(self.nrows):
# Fill rows with defaults
self.table.row.append()
self.table.flush()
if self.reopen:
self._reopen(mode='a')
self.table = self.h5file.root.table
def test01_attrs(self):
"""Checking indexing attributes (part1)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test01_attrs..." % self.__class__.__name__)
table = self.table
if self.iprops is DefaultProps:
self.assertEqual(table.indexed, 0)
else:
self.assertEqual(table.indexed, 1)
if self.iprops is DefaultProps:
self.assertEqual(table.colindexed["var1"], 0)
self.assertIsNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var2"], 0)
self.assertIsNone(table.cols.var2.index)
self.assertEqual(table.colindexed["var3"], 0)
self.assertIsNone(table.cols.var3.index)
self.assertEqual(table.colindexed["var4"], 0)
self.assertIsNone(table.cols.var4.index)
else:
# Check that the var1, var2 and var3 (and only these)
# has been indexed
self.assertEqual(table.colindexed["var1"], 1)
self.assertIsNotNone(table.cols.var1.index)
self.assertEqual(table.colindexed["var2"], 1)
self.assertIsNotNone(table.cols.var2.index)
self.assertEqual(table.colindexed["var3"], 1)
self.assertIsNotNone(table.cols.var3.index)
self.assertEqual(table.colindexed["var4"], 0)
self.assertIsNone(table.cols.var4.index)
def test02_attrs(self):
"""Checking indexing attributes (part2)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test02_attrs..." % self.__class__.__name__)
table = self.table
# Check the policy parameters
if common.verbose:
if table.indexed:
print("index props:", table.autoindex)
else:
print("Table is not indexed")
# Check non-default values for index saving policy
if self.iprops is NoAutoProps:
self.assertFalse(table.autoindex)
elif self.iprops is ChangeFiltersProps:
self.assertTrue(table.autoindex)
# Check Index() objects exists and are properly placed
if self.iprops is DefaultProps:
self.assertEqual(table.cols.var1.index, None)
self.assertEqual(table.cols.var2.index, None)
self.assertEqual(table.cols.var3.index, None)
self.assertEqual(table.cols.var4.index, None)
else:
self.assertIsInstance(table.cols.var1.index, tb.index.Index)
self.assertIsInstance(table.cols.var2.index, tb.index.Index)
self.assertIsInstance(table.cols.var3.index, tb.index.Index)
self.assertEqual(table.cols.var4.index, None)
def test03_counters(self):
"""Checking indexing counters"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test03_counters..." % self.__class__.__name__)
table = self.table
# Check the counters for indexes
if common.verbose:
if table.indexed:
print("indexedrows:", table._indexedrows)
print("unsavedindexedrows:", table._unsaved_indexedrows)
index = table.cols.var1.index
print("table rows:", table.nrows)
print("computed indexed rows:", index.nrows * index.slicesize)
else:
print("Table is not indexed")
if self.iprops is not DefaultProps:
index = table.cols.var1.index
indexedrows = index.nelements
self.assertEqual(table._indexedrows, indexedrows)
indexedrows = index.nelements
self.assertEqual(table._unsaved_indexedrows,
self.nrows - indexedrows)
def test04_noauto(self):
"""Checking indexing counters (non-automatic mode)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test04_noauto..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Check the counters for indexes
if common.verbose:
if table.indexed:
print("indexedrows:", table._indexedrows)
print("unsavedindexedrows:", table._unsaved_indexedrows)
index = table.cols.var1.index
print("computed indexed rows:", index.nelements)
else:
print("Table is not indexed")
# No unindexated rows should remain
index = table.cols.var1.index
if self.iprops is DefaultProps:
self.assertIsNone(index)
else:
indexedrows = index.nelements
self.assertEqual(table._indexedrows, index.nelements)
self.assertEqual(table._unsaved_indexedrows,
self.nrows - indexedrows)
# Check non-default values for index saving policy
if self.iprops is NoAutoProps:
self.assertFalse(table.autoindex)
elif self.iprops is ChangeFiltersProps:
self.assertTrue(table.autoindex)
def test05_icounters(self):
"""Checking indexing counters (remove_rows)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test05_icounters..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Non indexated rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
unsavedindexedrows = table._unsaved_indexedrows
# Now, remove some rows:
table.remove_rows(2, 4)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Check the counters for indexes
if common.verbose:
if table.indexed:
print("indexedrows:", table._indexedrows)
print("original indexedrows:", indexedrows)
print("unsavedindexedrows:", table._unsaved_indexedrows)
print("original unsavedindexedrows:", unsavedindexedrows)
# index = table.cols.var1.index
print("index dirty:", table.cols.var1.index.dirty)
else:
print("Table is not indexed")
# Check the counters
self.assertEqual(table.nrows, self.nrows - 2)
if self.iprops is NoAutoProps:
self.assertTrue(table.cols.var1.index.dirty)
# Check non-default values for index saving policy
if self.iprops is NoAutoProps:
self.assertFalse(table.autoindex)
elif self.iprops is ChangeFiltersProps:
self.assertTrue(table.autoindex)
def test06_dirty(self):
"""Checking dirty flags (remove_rows action)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test06_dirty..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Now, remove some rows:
table.remove_rows(3, 5)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Check the dirty flag for indexes
if common.verbose:
print("auto flag:", table.autoindex)
for colname in table.colnames:
if table.cols._f_col(colname).index:
print("dirty flag col %s: %s" %
(colname, table.cols._f_col(colname).index.dirty))
# Check the flags
for colname in table.colnames:
if table.cols._f_col(colname).index:
if not table.autoindex:
self.assertEqual(table.cols._f_col(colname).index.dirty,
True)
else:
self.assertEqual(table.cols._f_col(colname).index.dirty,
False)
def test07_noauto(self):
"""Checking indexing counters (modify_rows, no-auto mode)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test07_noauto..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# No unindexated rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
unsavedindexedrows = table._unsaved_indexedrows
# Now, modify just one row:
table.modify_rows(3, None, 1, [("asa", 0, 3, 3.1)])
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Check the counters for indexes
if common.verbose:
if table.indexed:
print("indexedrows:", table._indexedrows)
print("original indexedrows:", indexedrows)
print("unsavedindexedrows:", table._unsaved_indexedrows)
print("original unsavedindexedrows:", unsavedindexedrows)
index = table.cols.var1.index
print("computed indexed rows:", index.nelements)
else:
print("Table is not indexed")
# Check the counters
self.assertEqual(table.nrows, self.nrows)
if self.iprops is NoAutoProps:
self.assertTrue(table.cols.var1.index.dirty)
# Check the dirty flag for indexes
if common.verbose:
for colname in table.colnames:
if table.cols._f_col(colname).index:
print("dirty flag col %s: %s" %
(colname, table.cols._f_col(colname).index.dirty))
for colname in table.colnames:
if table.cols._f_col(colname).index:
if not table.autoindex:
self.assertEqual(table.cols._f_col(colname).index.dirty,
True)
else:
self.assertEqual(table.cols._f_col(colname).index.dirty,
False)
def test07b_noauto(self):
"""Checking indexing queries (modify in iterator, no-auto mode)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test07b_noauto..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Do a query that uses indexes
res = [row.nrow for row in table.where('(var2 == True) & (var3 > 0)')]
# Now, modify just one row:
for row in table:
if row.nrow == 3:
row['var1'] = "asa"
row['var2'] = True
row['var3'] = 3
row['var4'] = 3.1
row.update()
table.flush()
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Do a query that uses indexes
resq = [row.nrow for row in table.where('(var2 == True) & (var3 > 0)')]
res_ = res + [3]
if common.verbose:
print("AutoIndex?:", table.autoindex)
print("Query results (original):", res)
print("Query results (after modifying table):", resq)
print("Should look like:", res_)
self.assertEqual(res_, resq)
def test07c_noauto(self):
"""Checking indexing queries (append, no-auto mode)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test07c_noauto..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Do a query that uses indexes
res = [row.nrow for row in table.where('(var2 == True) & (var3 > 0)')]
# Now, append three rows
table.append([("asa", True, 1, 3.1)])
table.append([("asb", True, 2, 3.1)])
table.append([("asc", True, 3, 3.1)])
table.flush()
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Do a query that uses indexes
resq = [row.nrow for row in table.where('(var2 == True) & (var3 > 0)')]
res_ = res + [table.nrows-3, table.nrows-2, table.nrows-1]
if common.verbose:
print("AutoIndex?:", table.autoindex)
print("Query results (original):", res)
print("Query results (after modifying table):", resq)
print("Should look like:", res_)
self.assertEqual(res_, resq)
def test08_dirty(self):
"""Checking dirty flags (modify_columns)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test08_dirty..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Non indexed rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
self.assertIsNotNone(indexedrows)
unsavedindexedrows = table._unsaved_indexedrows
self.assertIsNotNone(unsavedindexedrows)
# Now, modify a couple of rows:
table.modify_columns(1, columns=[["asa", "asb"], [1., 2.]],
names=["var1", "var4"])
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
# Check the counters
self.assertEqual(table.nrows, self.nrows)
if self.iprops is NoAutoProps:
self.assertTrue(table.cols.var1.index.dirty)
# Check the dirty flag for indexes
if common.verbose:
for colname in table.colnames:
if table.cols._f_col(colname).index:
print("dirty flag col %s: %s" %
(colname, table.cols._f_col(colname).index.dirty))
for colname in table.colnames:
if table.cols._f_col(colname).index:
if not table.autoindex:
if colname in ["var1"]:
self.assertEqual(
table.cols._f_col(colname).index.dirty, True)
else:
self.assertEqual(
table.cols._f_col(colname).index.dirty, False)
else:
self.assertEqual(table.cols._f_col(colname).index.dirty,
False)
def test09a_propIndex(self):
"""Checking propagate Index feature in Table.copy() (attrs)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test09a_propIndex..." % self.__class__.__name__)
table = self.table
# Don't force a sync in indexes
# table.flush_rows_to_index()
# Non indexed rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
self.assertIsNotNone(indexedrows)
unsavedindexedrows = table._unsaved_indexedrows
self.assertIsNotNone(unsavedindexedrows)
# Now, remove some rows to make columns dirty
# table.remove_rows(3,5)
# Copy a Table to another location
table2 = table.copy("/", 'table2', propindexes=True)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
table2 = self.h5file.root.table2
index1 = table.cols.var1.index
index2 = table2.cols.var1.index
if common.verbose:
print("Copied index:", index2)
print("Original index:", index1)
if index1:
print("Elements in copied index:", index2.nelements)
print("Elements in original index:", index1.nelements)
# Check the counters
self.assertEqual(table.nrows, table2.nrows)
if table.indexed:
self.assertTrue(table2.indexed)
if self.iprops is DefaultProps:
# No index: the index should not exist
self.assertIsNone(index1)
self.assertIsNone(index2)
elif self.iprops is NoAutoProps:
self.assertIsNotNone(index2)
# Check the dirty flag for indexes
if common.verbose:
for colname in table2.colnames:
if table2.cols._f_col(colname).index:
print("dirty flag col %s: %s" %
(colname, table2.cols._f_col(colname).index.dirty))
for colname in table2.colnames:
if table2.cols._f_col(colname).index:
self.assertEqual(table2.cols._f_col(colname).index.dirty,
False)
def test09b_propIndex(self):
"""Checking that propindexes=False works"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test09b_propIndex..." % self.__class__.__name__)
table = self.table
# Don't force a sync in indexes
# table.flush_rows_to_index()
# Non indexed rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
self.assertIsNotNone(indexedrows)
unsavedindexedrows = table._unsaved_indexedrows
self.assertIsNotNone(unsavedindexedrows)
# Now, remove some rows to make columns dirty
# table.remove_rows(3,5)
# Copy a Table to another location
table2 = table.copy("/", 'table2', propindexes=False)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
table2 = self.h5file.root.table2
if common.verbose:
print("autoindex?:", self.iprops.auto)
print("Copied index indexed?:", table2.cols.var1.is_indexed)
print("Original index indexed?:", table.cols.var1.is_indexed)
if self.iprops is DefaultProps:
# No index: the index should not exist
self.assertFalse(table2.cols.var1.is_indexed)
self.assertFalse(table.cols.var1.is_indexed)
elif self.iprops is NoAutoProps:
self.assertFalse(table2.cols.var1.is_indexed)
self.assertTrue(table.cols.var1.is_indexed)
def test10_propIndex(self):
"""Checking propagate Index feature in Table.copy() (values)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test10_propIndex..." % self.__class__.__name__)
table = self.table
# Don't force a sync in indexes
# table.flush_rows_to_index()
# Non indexed rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
self.assertIsNotNone(indexedrows)
unsavedindexedrows = table._unsaved_indexedrows
self.assertIsNotNone(unsavedindexedrows)
# Now, remove some rows to make columns dirty
# table.remove_rows(3,5)
# Copy a Table to another location
table2 = table.copy("/", 'table2', propindexes=True)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
table2 = self.h5file.root.table2
index1 = table.cols.var3.index
index2 = table2.cols.var3.index
if common.verbose:
print("Copied index:", index2)
print("Original index:", index1)
if index1:
print("Elements in copied index:", index2.nelements)
print("Elements in original index:", index1.nelements)
def test11_propIndex(self):
"""Checking propagate Index feature in Table.copy() (dirty flags)"""
if common.verbose:
print('\n', '-=' * 30)
print("Running %s.test11_propIndex..." % self.__class__.__name__)
table = self.table
# Force a sync in indexes
table.flush_rows_to_index()
# Non indexed rows should remain here
if self.iprops is not DefaultProps:
indexedrows = table._indexedrows
self.assertIsNotNone(indexedrows)
unsavedindexedrows = table._unsaved_indexedrows
self.assertIsNotNone(unsavedindexedrows)
# Now, modify an indexed column and an unindexed one
# to make the "var1" dirty
table.modify_columns(1, columns=[["asa", "asb"], [1., 2.]],
names=["var1", "var4"])
# Copy a Table to another location
table2 = table.copy("/", 'table2', propindexes=True)
if self.reopen:
self._reopen(mode='a')
table = self.h5file.root.table
table2 = self.h5file.root.table2
index1 = table.cols.var1.index
index2 = table2.cols.var1.index
if common.verbose:
print("Copied index:", index2)
print("Original index:", index1)
if index1:
print("Elements in copied index:", index2.nelements)
print("Elements in original index:", index1.nelements)
# Check the dirty flag for indexes
if common.verbose:
for colname in table2.colnames:
if table2.cols._f_col(colname).index:
print("dirty flag col %s: %s" %
(colname, table2.cols._f_col(colname).index.dirty))
for colname in table2.colnames:
if table2.cols._f_col(colname).index:
if table2.autoindex:
# All the destination columns should be non-dirty because
# the copy removes the dirty state and puts the
# index in a sane state
self.assertEqual(table2.cols._f_col(colname).index.dirty,
False)
# minRowIndex = 10000 # just if one wants more indexed rows to be checked
class AI1TestCase(AutomaticIndexingTestCase):
# nrows = 10002
nrows = 102
reopen = 0
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI2TestCase(AutomaticIndexingTestCase):
# nrows = 10002
nrows = 102
reopen = 1
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI4bTestCase(AutomaticIndexingTestCase):
# nrows = 10012
nrows = 112
reopen = 1
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI5TestCase(AutomaticIndexingTestCase):
sbs, bs, ss, cs = tb.idxutils.calc_chunksize(minRowIndex, memlevel=1)
nrows = ss * 11-1
reopen = 0
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI6TestCase(AutomaticIndexingTestCase):
sbs, bs, ss, cs = tb.idxutils.calc_chunksize(minRowIndex, memlevel=1)
nrows = ss * 21 + 1
reopen = 1
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI7TestCase(AutomaticIndexingTestCase):
sbs, bs, ss, cs = tb.idxutils.calc_chunksize(minRowIndex, memlevel=1)
nrows = ss * 12-1
# nrows = ss * 1-1 # faster test
reopen = 0
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI8TestCase(AutomaticIndexingTestCase):
sbs, bs, ss, cs = tb.idxutils.calc_chunksize(minRowIndex, memlevel=1)
nrows = ss * 15 + 100
# nrows = ss * 1 + 100 # faster test
reopen = 1
iprops = NoAutoProps
colsToIndex = ['var1', 'var2', 'var3']
class AI9TestCase(AutomaticIndexingTestCase):
sbs, bs, ss, cs = tb.idxutils.calc_chunksize(minRowIndex, memlevel=1)
nrows = ss
reopen = 0
iprops = DefaultProps
colsToIndex = []
class AI10TestCase(AutomaticIndexingTestCase):
# nrows = 10002
nrows = 102
reopen = 1
iprops = DefaultProps
colsToIndex = []
class AI11TestCase(AutomaticIndexingTestCase):
# nrows = 10002
nrows = 102
reopen = 0
iprops = ChangeFiltersProps
colsToIndex = ['var1', 'var2', 'var3']
class AI12TestCase(AutomaticIndexingTestCase):
# nrows = 10002
nrows = 102
reopen = 0
iprops = ChangeFiltersProps
colsToIndex = ['var1', 'var2', 'var3']
class ManyNodesTestCase(common.TempFileMixin, common.PyTablesTestCase):
opem_kwargs = dict(node_cache_slots=64)
def test00(self):
"""Indexing many nodes in one single session (based on bug #26)"""
IdxRecord = {
'f0': tb.Int8Col(),
'f1': tb.Int8Col(),
'f2': tb.Int8Col(),
}
for qn in range(5):
for sn in range(5):
qchr = 'chr' + str(qn)
name = 'chr' + str(sn)
path = "/at/%s/pt" % (qchr)
table = self.h5file.create_table(
path, name, IdxRecord, createparents=1)
table.cols.f0.create_index()
table.cols.f1.create_index()
table.cols.f2.create_index()
table.row.append()
table.flush()
class IndexPropsChangeTestCase(common.TempFileMixin, common.PyTablesTestCase):
"""Test case for changing index properties in a table."""
class MyDescription(tb.IsDescription):
icol = tb.IntCol()
oldIndexProps = IndexProps()
newIndexProps = IndexProps(auto=False, filters=tb.Filters(complevel=9))
def setUp(self):
super().setUp()
table = self.h5file.create_table('/', 'test', self.MyDescription)
table.autoindex = self.oldIndexProps.auto
row = table.row
for i in range(100):
row['icol'] = i % 25
row.append()
table.flush()
self.table = table
def test_attributes(self):
"""Storing index properties as table attributes."""
for refprops in [self.oldIndexProps, self.newIndexProps]:
self.assertEqual(self.table.autoindex, refprops.auto)
self.table.autoindex = self.newIndexProps.auto
def test_copyattrs(self):
"""Copying index properties attributes."""
oldtable = self.table
newtable = oldtable.copy('/', 'test2')
self.assertEqual(oldtable.autoindex, newtable.autoindex)
class IndexFiltersTestCase(common.TempFileMixin, common.PyTablesTestCase):
"""Test case for setting index filters."""
def setUp(self):
super().setUp()
description = {'icol': tb.IntCol()}
self.table = self.h5file.create_table('/', 'test', description)
def test_createIndex(self):
"""Checking input parameters in new indexes."""
# Different from default.
argfilters = copy.copy(tb.index.default_index_filters)
argfilters.shuffle = not tb.index.default_index_filters.shuffle
# Different both from default and the previous one.
idxfilters = copy.copy(tb.index.default_index_filters)
idxfilters.shuffle = not tb.index.default_index_filters.shuffle
idxfilters.fletcher32 = not tb.index.default_index_filters.fletcher32
icol = self.table.cols.icol
# First create
icol.create_index(kind='ultralight', optlevel=4)
self.assertEqual(icol.index.kind, 'ultralight')
self.assertEqual(icol.index.optlevel, 4)
self.assertEqual(icol.index.filters, tb.index.default_index_filters)
icol.remove_index()
# Second create
icol.create_index(kind='medium', optlevel=3, filters=argfilters)
self.assertEqual(icol.index.kind, 'medium')
self.assertEqual(icol.index.optlevel, 3)
self.assertEqual(icol.index.filters, argfilters)
icol.remove_index()
def test_reindex(self):
"""Checking input parameters in recomputed indexes."""
icol = self.table.cols.icol
icol.create_index(
kind='full', optlevel=5, filters=tb.Filters(complevel=3))
kind = icol.index.kind
optlevel = icol.index.optlevel
filters = icol.index.filters
icol.reindex()
ni = icol.index
if common.verbose:
print(f"Old parameters: {kind}, {optlevel}, {filters}")
print("New parameters: {}, {}, {}".format(
ni.kind, ni.optlevel, ni.filters))
self.assertEqual(ni.kind, kind)
self.assertEqual(ni.optlevel, optlevel)
self.assertEqual(ni.filters, filters)
class OldIndexTestCase(common.TestFileMixin, common.PyTablesTestCase):
h5fname = common.test_filename("idx-std-1.x.h5")
def test1_x(self):
"""Check that files with 1.x indexes are recognized and warned."""
self.assertWarns(tb.exceptions.OldIndexWarning,
self.h5file.get_node, "/table")
# Sensible parameters for indexing with small blocksizes
small_blocksizes = (512, 128, 32, 8)
class CompletelySortedIndexTestCase(common.TempFileMixin,
common.PyTablesTestCase):
"""Test case for testing a complete sort in a table."""
nrows = 100
nrowsinbuf = 11
class MyDescription(tb.IsDescription):
rcol = tb.IntCol(pos=1)
icol = tb.IntCol(pos=2)
def setUp(self):
super().setUp()
table = self.h5file.create_table('/', 'table', self.MyDescription)
row = table.row
nrows = self.nrows
for i in range(nrows):
row['rcol'] = i
row['icol'] = nrows - i
row.append()
table.flush()
self.table = table
self.icol = self.table.cols.icol
# A full index with maximum optlevel should always be completely sorted
self.icol.create_csindex(_blocksizes=small_blocksizes)
def test00_isCompletelySortedIndex(self):
"""Testing the Column.is_csi property."""
icol = self.icol
self.assertEqual(icol.index.is_csi, True)
icol.remove_index()
# Other kinds than full, should never return a CSI
icol.create_index(kind="medium", optlevel=9)
self.assertEqual(icol.index.is_csi, False)
icol.remove_index()
# As the table is small, lesser optlevels should be able to
# create a completely sorted index too.
icol.create_index(kind="full", optlevel=6)
self.assertEqual(icol.index.is_csi, True)
# Checking a CSI in a sorted copy
self.table.copy("/", 'table2', sortby='icol', checkCSI=True)
self.assertEqual(icol.index.is_csi, True)
def test01_readSorted1(self):
"""Testing the Index.read_sorted() method with no arguments."""
icol = self.icol
sortedcol = np.sort(icol[:])
sortedcol2 = icol.index.read_sorted()
if common.verbose:
print("Original sorted column:", sortedcol)
print("The values from the index:", sortedcol2)
self.assertTrue(common.allequal(sortedcol, sortedcol2))
def test01_readSorted2(self):
"""Testing the Index.read_sorted() method with arguments (I)."""
icol = self.icol
sortedcol = np.sort(icol[:])[30:55]
sortedcol2 = icol.index.read_sorted(30, 55)
if common.verbose:
print("Original sorted column:", sortedcol)
print("The values from the index:", sortedcol2)
self.assertTrue(common.allequal(sortedcol, sortedcol2))
def test01_readSorted3(self):
"""Testing the Index.read_sorted() method with arguments (II)."""
icol = self.icol
sortedcol = np.sort(icol[:])[33:97]
sortedcol2 = icol.index.read_sorted(33, 97)
if common.verbose:
print("Original sorted column:", sortedcol)
print("The values from the index:", sortedcol2)
self.assertTrue(common.allequal(sortedcol, sortedcol2))
def test02_readIndices1(self):
"""Testing the Index.read_indices() method with no arguments."""
icol = self.icol
indicescol = np.argsort(icol[:]).astype('uint64')
indicescol2 = icol.index.read_indices()
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test02_readIndices2(self):
"""Testing the Index.read_indices() method with arguments (I)."""
icol = self.icol
indicescol = np.argsort(icol[:])[30:55].astype('uint64')
indicescol2 = icol.index.read_indices(30, 55)
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test02_readIndices3(self):
"""Testing the Index.read_indices() method with arguments (II)."""
icol = self.icol
indicescol = np.argsort(icol[:])[33:97].astype('uint64')
indicescol2 = icol.index.read_indices(33, 97)
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test02_readIndices4(self):
"""Testing the Index.read_indices() method with arguments (III)."""
icol = self.icol
indicescol = np.argsort(icol[:])[33:97:2].astype('uint64')
indicescol2 = icol.index.read_indices(33, 97, 2)
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test02_readIndices5(self):
"""Testing the Index.read_indices() method with arguments (IV)."""
icol = self.icol
indicescol = np.argsort(icol[:])[33:55:5].astype('uint64')
indicescol2 = icol.index.read_indices(33, 55, 5)
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test02_readIndices6(self):
"""Testing the Index.read_indices() method with step only."""
icol = self.icol
indicescol = np.argsort(icol[:])[::3].astype('uint64')
indicescol2 = icol.index.read_indices(step=3)
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test03_getitem1(self):
"""Testing the Index.__getitem__() method with no arguments."""
icol = self.icol
indicescol = np.argsort(icol[:]).astype('uint64')
indicescol2 = icol.index[:]
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test03_getitem2(self):
"""Testing the Index.__getitem__() method with start."""
icol = self.icol
indicescol = np.argsort(icol[:])[31].astype('uint64')
indicescol2 = icol.index[31]
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test03_getitem3(self):
"""Testing the Index.__getitem__() method with start, stop."""
icol = self.icol
indicescol = np.argsort(icol[:])[2:16].astype('uint64')
indicescol2 = icol.index[2:16]
if common.verbose:
print("Original indices column:", indicescol)
print("The values from the index:", indicescol2)
self.assertTrue(common.allequal(indicescol, indicescol2))
def test04_itersorted1(self):
"""Testing the Table.itersorted() method with no arguments."""
table = self.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol')], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted2(self):
"""Testing the Table.itersorted() method with a start."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[15:]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', start=15)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted3(self):
"""Testing the Table.itersorted() method with a stop."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[:20]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', stop=20)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted4(self):
"""Testing the Table.itersorted() method with a start and stop."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[15:20]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', start=15, stop=20)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted5(self):
"""Testing the Table.itersorted() method with a start, stop and
step."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[15:45:4]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', start=15, stop=45, step=4)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted6(self):
"""Testing the Table.itersorted() method with a start, stop and
step."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[33:55:5]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', start=33, stop=55, step=5)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted7(self):
"""Testing the Table.itersorted() method with checkCSI=True."""
table = self.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', checkCSI=True)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted8(self):
"""Testing the Table.itersorted() method with a start, stop and
negative step."""
# see also gh-252
table = self.table
sortedtable = np.sort(table[:], order='icol')[55:33:-5]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', start=55, stop=33, step=-5)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted9(self):
"""Testing the Table.itersorted() method with a negative step -5."""
# see also gh-252
table = self.table
sortedtable = np.sort(table[:], order='icol')[::-5]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', step=-5)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test04_itersorted10(self):
"""Testing the Table.itersorted() method with a negative step -1."""
# see also gh-252
table = self.table
sortedtable = np.sort(table[:], order='icol')[::-1]
sortedtable2 = np.array(
[row.fetch_all_fields() for row in table.itersorted(
'icol', step=-1)], dtype=table._v_dtype)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from the iterator:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted1(self):
"""Testing the Table.read_sorted() method with no arguments."""
table = self.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table.read_sorted('icol')
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted2(self):
"""Testing the Table.read_sorted() method with a start."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[16:17]
sortedtable2 = table.read_sorted('icol', start=16)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted3(self):
"""Testing the Table.read_sorted() method with a start and stop."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[16:33]
sortedtable2 = table.read_sorted('icol', start=16, stop=33)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted4(self):
"""Testing the Table.read_sorted() method with a start, stop and
step."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[33:55:5]
sortedtable2 = table.read_sorted('icol', start=33, stop=55, step=5)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted5(self):
"""Testing the Table.read_sorted() method with only a step."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[::3]
sortedtable2 = table.read_sorted('icol', step=3)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted6(self):
"""Testing the Table.read_sorted() method with negative step."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[::-1]
sortedtable2 = table.read_sorted('icol', step=-1)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted7(self):
"""Testing the Table.read_sorted() method with negative step (II)."""
table = self.table
sortedtable = np.sort(table[:], order='icol')[::-2]
sortedtable2 = table.read_sorted('icol', step=-2)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted8(self):
"""Testing the Table.read_sorted() method with negative step (III))."""
table = self.table
sstart = 100-24-1
sstop = 100-54-1
sortedtable = np.sort(table[:], order='icol')[sstart:sstop:-1]
sortedtable2 = table.read_sorted('icol', start=24, stop=54, step=-1)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted9(self):
"""Testing the Table.read_sorted() method with negative step (IV))."""
table = self.table
sstart = 100-14-1
sstop = 100-54-1
sortedtable = np.sort(table[:], order='icol')[sstart:sstop:-3]
sortedtable2 = table.read_sorted('icol', start=14, stop=54, step=-3)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted10(self):
"""Testing the Table.read_sorted() method with negative step (V))."""
table = self.table
sstart = 100-24-1
sstop = 100-25-1
sortedtable = np.sort(table[:], order='icol')[sstart:sstop:-2]
sortedtable2 = table.read_sorted('icol', start=24, stop=25, step=-2)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05_readSorted11(self):
"""Testing the Table.read_sorted() method with start > stop."""
table = self.table
sstart = 100-137-1
sstop = 100-25-1
sortedtable = np.sort(table[:], order='icol')[sstart:sstop:-2]
sortedtable2 = table.read_sorted('icol', start=137, stop=25, step=-2)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05a_readSorted12(self):
"""Testing the Table.read_sorted() method with checkCSI (I)."""
table = self.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table.read_sorted('icol', checkCSI=True)
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test05b_readSorted12(self):
"""Testing the Table.read_sorted() method with checkCSI (II)."""
table = self.table
self.assertRaises(ValueError,
table.read_sorted, "rcol", checkCSI=False)
def test06_copy_sorted1(self):
"""Testing the Table.copy(sortby) method with no arguments."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted2(self):
"""Testing the Table.copy(sortby) method with step=-1."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol", step=-1)
sortedtable = np.sort(table[:], order='icol')[::-1]
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted3(self):
"""Testing the Table.copy(sortby) method with only a start."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol", start=3)
sortedtable = np.sort(table[:], order='icol')[3:4]
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted4(self):
"""Testing the Table.copy(sortby) method with start, stop."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol", start=3, stop=40)
sortedtable = np.sort(table[:], order='icol')[3:40]
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted5(self):
"""Testing the Table.copy(sortby) method with start, stop, step."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol",
start=3, stop=33, step=5)
sortedtable = np.sort(table[:], order='icol')[3:33:5]
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted6(self):
"""Testing the Table.copy(sortby) method after table re-opening."""
self._reopen(mode='a')
table = self.h5file.root.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table2[:]
if common.verbose:
print("Original sorted table:", sortedtable)
print("The values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test06_copy_sorted7(self):
"""Testing the `checkCSI` parameter of Table.copy() (I)."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
self.assertRaises(ValueError,
table2.copy, "/", 'table3',
sortby="rcol", checkCSI=False)
def test06_copy_sorted8(self):
"""Testing the `checkCSI` parameter of Table.copy() (II)."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
self.assertRaises(ValueError,
table2.copy, "/", 'table3',
sortby="rcol", checkCSI=True)
def test07_isCSI_noelements(self):
"""Testing the representation of an index with no elements."""
t2 = self.h5file.create_table('/', 't2', self.MyDescription)
irows = t2.cols.rcol.create_csindex()
if common.verbose:
print("repr(t2)-->\n", repr(t2))
self.assertEqual(irows, 0)
self.assertEqual(t2.colindexes['rcol'].is_csi, False)
class ReadSortedIndexTestCase(common.TempFileMixin, common.PyTablesTestCase):
"""Test case for testing sorted reading in a "full" sorted column."""
nrows = 100
nrowsinbuf = 11
class MyDescription(tb.IsDescription):
rcol = tb.IntCol(pos=1)
icol = tb.IntCol(pos=2)
def setUp(self):
super().setUp()
table = self.h5file.create_table('/', 'table', self.MyDescription)
row = table.row
nrows = self.nrows
for i in range(nrows):
row['rcol'] = i
row['icol'] = nrows - i
row.append()
table.flush()
self.table = table
self.icol = self.table.cols.icol
# A full index with maximum optlevel should always be completely sorted
self.icol.create_index(optlevel=self.optlevel, kind="full",
_blocksizes=small_blocksizes)
def test01_readSorted1(self):
"""Testing the Table.read_sorted() method with no arguments."""
table = self.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table.read_sorted('icol')
if common.verbose:
print("Sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
# Compare with the sorted read table because we have no
# guarantees that read_sorted returns a completely sorted table
self.assertTrue(common.allequal(
sortedtable, np.sort(sortedtable2, order="icol")))
def test01_readSorted2(self):
"""Testing the Table.read_sorted() method with no arguments
(re-open)."""
self._reopen()
table = self.h5file.root.table
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = table.read_sorted('icol')
if common.verbose:
print("Sorted table:", sortedtable)
print("The values from read_sorted:", sortedtable2)
# Compare with the sorted read table because we have no
# guarantees that read_sorted returns a completely sorted table
self.assertTrue(common.allequal(
sortedtable, np.sort(sortedtable2, order="icol")))
def test02_copy_sorted1(self):
"""Testing the Table.copy(sortby) method."""
table = self.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = np.sort(table2[:], order='icol')
if common.verbose:
print("Original table:", table2[:])
print("The sorted values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
def test02_copy_sorted2(self):
"""Testing the Table.copy(sortby) method after table re-opening."""
self._reopen(mode='a')
table = self.h5file.root.table
# Copy to another table
table.nrowsinbuf = self.nrowsinbuf
table2 = table.copy("/", 'table2', sortby="icol")
sortedtable = np.sort(table[:], order='icol')
sortedtable2 = np.sort(table2[:], order='icol')
if common.verbose:
print("Original table:", table2[:])
print("The sorted values from copy:", sortedtable2)
self.assertTrue(common.allequal(sortedtable, sortedtable2))
class ReadSortedIndex0(ReadSortedIndexTestCase):
optlevel = 0
class ReadSortedIndex3(ReadSortedIndexTestCase):
optlevel = 3
class ReadSortedIndex6(ReadSortedIndexTestCase):
optlevel = 6
class ReadSortedIndex9(ReadSortedIndexTestCase):
optlevel = 9
class Issue156TestBase(common.TempFileMixin, common.PyTablesTestCase):
# field name in table according to which test_copysort() sorts the table
sort_field = None
def setUp(self):
super().setUp()
# create nested table
class Foo(tb.IsDescription):
frame = tb.UInt16Col()
class Bar(tb.IsDescription):
code = tb.UInt16Col()
table = self.h5file.create_table('/', 'foo', Foo,
filters=tb.Filters(3, 'zlib'),
createparents=True)
self.h5file.flush()
# fill table with 10 random numbers
for k in range(10):
row = table.row
row['frame'] = np.random.randint(0, 2**16-1)
row['Bar/code'] = np.random.randint(0, 2**16-1)
row.append()
self.h5file.flush()
def test_copysort(self):
# copy table
oldNode = self.h5file.get_node('/foo')
# create completely sorted index on a main column
oldNode.colinstances[self.sort_field].create_csindex()
# this fails on ade2ba123efd267fd31
# see gh-156
new_node = oldNode.copy(newname='foo2', overwrite=True,
sortby=self.sort_field, checkCSI=True,
propindexes=True)
# check column is sorted
self.assertTrue(np.all(
new_node.col(self.sort_field) ==
sorted(oldNode.col(self.sort_field))))
# check index is available
self.assertIn(self.sort_field, new_node.colindexes)
# check CSI was propagated
self.assertTrue(new_node.colindexes[self.sort_field].is_csi)
class Issue156TestCase01(Issue156TestBase):
# sort by field from non nested entry
sort_field = 'frame'
class Issue156TestCase02(Issue156TestBase):
# sort by field from nested entry
sort_field = 'Bar/code'
class Issue119Time32ColTestCase(common.TempFileMixin, common.PyTablesTestCase):
"""TimeCol not properly indexing."""
col_typ = tb.Time32Col
values = [
0.93240451618785880,
0.76322375510776170,
0.16695030056300875,
0.91259117097807850,
0.93977847053454630,
0.51450406513503090,
0.24452129962257563,
0.85475938924825230,
0.32512326762476930,
0.75127635627046820,
]
def setUp(self):
super().setUp()
class Descr(tb.IsDescription):
when = self.col_typ(pos=1)
value = tb.Float32Col(pos=2)
self.table = self.h5file.create_table('/', 'test', Descr)
self.t = 1321031471.0 # 11/11/11 11:11:11
data = [(self.t + i, item) for i, item in enumerate(self.values)]
self.table.append(data)
self.h5file.flush()
def test_timecol_issue(self):
tbl = self.table
t = self.t
wherestr = '(when >= %d) & (when < %d)' % (t, t + 5)
no_index = tbl.read_where(wherestr)
tbl.cols.when.create_index(_verbose=False)
with_index = tbl.read_where(wherestr)
self.assertTrue((no_index == with_index).all())
class Issue119Time64ColTestCase(Issue119Time32ColTestCase):
col_typ = tb.Time64Col
class TestIndexingNans(common.TempFileMixin, common.PyTablesTestCase):
def test_issue_282(self):
trMap = {'index': tb.Int64Col(), 'values': tb.FloatCol()}
table = self.h5file.create_table('/', 'table', trMap)
r = table.row
for i in range(5):
r['index'] = i
r['values'] = np.nan if i == 0 else i
r.append()
table.flush()
table.cols.values.create_index()
# retrieve
result = table.read_where('(values >= 0)')
self.assertEqual(len(result), 4)
def test_issue_327(self):
table = self.h5file.create_table('/', 'table', dict(
index=tb.Int64Col(),
values=tb.FloatCol(shape=()),
values2=tb.FloatCol(shape=()),
))
r = table.row
for i in range(5):
r['index'] = i
r['values'] = np.nan if i == 2 or i == 3 else i
r['values2'] = i
r.append()
table.flush()
table.cols.values.create_index()
table.cols.values2.create_index()
results2 = table.read_where('(values2 > 0)')
self.assertEqual(len(results2), 4)
results = table.read_where('(values > 0)')
self.assertEqual(len(results), 2)
def test_issue_327_b(self):
table = self.h5file.create_table('/', 'table', dict(
index=tb.Int64Col(),
values=tb.FloatCol(shape=()),
values2=tb.FloatCol(shape=()),
))
r = table.row
for _ in range(100):
for i in range(5):
r['index'] = i
r['values'] = np.nan if i == 2 or i == 3 else i
r['values2'] = i
r.append()
table.flush()
table.cols.values.create_index(_blocksizes=small_blocksizes)
table.cols.values2.create_index(_blocksizes=small_blocksizes)
results2 = table.read_where('(values2 > 0)')
self.assertEqual(len(results2), 400)
results = table.read_where('(values > 0)')
self.assertEqual(len(results), 200)
def test_csindex_nans(self):
table = self.h5file.create_table('/', 'table', dict(
index=tb.Int64Col(),
values=tb.FloatCol(shape=()),
values2=tb.FloatCol(shape=()),
))
r = table.row
for x in range(100):
for i in range(5):
r['index'] = i
r['values'] = np.nan if i == 2 or i == 3 else i
r['values2'] = i
r.append()
table.flush()
table.cols.values.create_csindex(_blocksizes=small_blocksizes)
table.cols.values2.create_csindex(_blocksizes=small_blocksizes)
results2 = table.read_where('(values2 > 0)')
self.assertEqual(len(results2), 100*4)
results = table.read_where('(values > 0)')
self.assertEqual(len(results), 100*2)
def suite():
theSuite = common.unittest.TestSuite()
niter = 1
# heavy = 1 # Uncomment this only for testing purposes!
for n in range(niter):
theSuite.addTest(common.make_suite(BasicReadTestCase))
theSuite.addTest(common.make_suite(ZlibReadTestCase))
theSuite.addTest(common.make_suite(BloscReadTestCase))
theSuite.addTest(common.make_suite(LZOReadTestCase))
theSuite.addTest(common.make_suite(Bzip2ReadTestCase))
theSuite.addTest(common.make_suite(ShuffleReadTestCase))
theSuite.addTest(common.make_suite(Fletcher32ReadTestCase))
theSuite.addTest(
common.make_suite(ShuffleFletcher32ReadTestCase))
theSuite.addTest(common.make_suite(OneHalfTestCase))
theSuite.addTest(common.make_suite(UpperBoundTestCase))
theSuite.addTest(common.make_suite(LowerBoundTestCase))
theSuite.addTest(common.make_suite(AI1TestCase))
theSuite.addTest(common.make_suite(AI2TestCase))
theSuite.addTest(common.make_suite(AI9TestCase))
theSuite.addTest(common.make_suite(DeepTableIndexTestCase))
theSuite.addTest(common.make_suite(IndexPropsChangeTestCase))
theSuite.addTest(common.make_suite(IndexFiltersTestCase))
theSuite.addTest(common.make_suite(OldIndexTestCase))
theSuite.addTest(
common.make_suite(CompletelySortedIndexTestCase))
theSuite.addTest(common.make_suite(ManyNodesTestCase))
theSuite.addTest(common.make_suite(ReadSortedIndex0))
theSuite.addTest(common.make_suite(ReadSortedIndex3))
theSuite.addTest(common.make_suite(ReadSortedIndex6))
theSuite.addTest(common.make_suite(ReadSortedIndex9))
theSuite.addTest(common.make_suite(Issue156TestCase01))
theSuite.addTest(common.make_suite(Issue156TestCase02))
theSuite.addTest(common.make_suite(Issue119Time32ColTestCase))
theSuite.addTest(common.make_suite(Issue119Time64ColTestCase))
theSuite.addTest(common.make_suite(TestIndexingNans))
if common.heavy:
# These are too heavy for normal testing
theSuite.addTest(common.make_suite(AI4bTestCase))
theSuite.addTest(common.make_suite(AI5TestCase))
theSuite.addTest(common.make_suite(AI6TestCase))
theSuite.addTest(common.make_suite(AI7TestCase))
theSuite.addTest(common.make_suite(AI8TestCase))
theSuite.addTest(common.make_suite(AI10TestCase))
theSuite.addTest(common.make_suite(AI11TestCase))
theSuite.addTest(common.make_suite(AI12TestCase))
return theSuite
if __name__ == '__main__':
import sys
common.parse_argv(sys.argv)
common.print_versions()
common.unittest.main(defaultTest='suite')