@@ -63,37 +63,61 @@ public class AlignedSinglePageWholeChunkReader extends AbstractChunkReader
6363 // deleted intervals of all the sub sensors
6464 private final List <List <TimeRange >> valueDeleteIntervalsList = new ArrayList <>();
6565 private final long pageEstimatedMemoryUsageInBytes ;
66+ private final Chunk ownedTimeChunk ;
67+ private final List <Chunk > ownedValueChunks ;
6668
6769 public AlignedSinglePageWholeChunkReader (
6870 Chunk timeChunk , List <Chunk > valueChunkList , LongConsumer filteredRowsRecord )
6971 throws IOException {
7072 super (Long .MIN_VALUE , null , filteredRowsRecord );
73+ ownedTimeChunk = timeChunk ;
74+ ownedValueChunks = new ArrayList <>(valueChunkList );
7175 this .timeChunkHeader = timeChunk .getHeader ();
7276 this .timeChunkDataBuffer = timeChunk .getData ();
7377 this .encryptParam = timeChunk .getEncryptParam ();
74- this .pageEstimatedMemoryUsageInBytes =
75- calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData (timeChunk , valueChunkList );
76-
77- valueChunkList .forEach (
78- chunk -> {
79- this .valueChunkHeaderList .add (chunk == null ? null : chunk .getHeader ());
80- this .valueChunkDataBufferList .add (chunk == null ? null : chunk .getData ());
81- this .valueDeleteIntervalsList .add (chunk == null ? null : chunk .getDeleteIntervalList ());
82- });
78+ try {
79+ this .pageEstimatedMemoryUsageInBytes =
80+ calculateMaxPageEstimatedMemoryUsageInBytesWithBatchData (timeChunk , valueChunkList );
81+
82+ valueChunkList .forEach (
83+ chunk -> {
84+ this .valueChunkHeaderList .add (chunk == null ? null : chunk .getHeader ());
85+ this .valueChunkDataBufferList .add (chunk == null ? null : chunk .getData ());
86+ this .valueDeleteIntervalsList .add (chunk == null ? null : chunk .getDeleteIntervalList ());
87+ });
88+
89+ initAllPageReaders ();
90+ } catch (IOException | RuntimeException | Error e ) {
91+ try {
92+ close ();
93+ } catch (RuntimeException failure ) {
94+ e .addSuppressed (failure );
95+ }
96+ throw e ;
97+ }
98+ }
8399
84- initAllPageReaders ();
100+ @ Override
101+ public void close () {
102+ ownedTimeChunk .close ();
103+ for (Chunk chunk : ownedValueChunks ) {
104+ if (chunk != null ) {
105+ chunk .close ();
106+ }
107+ }
85108 }
86109
87110 private void initAllPageReaders () throws IOException {
111+ int pageIndex = 0 ;
88112 while (timeChunkDataBuffer .remaining () > 0 ) {
89- AlignedPageReader alignedPageReader = deserializeFromSinglePageChunk ();
113+ AlignedPageReader alignedPageReader = deserializeFromSinglePageChunk (pageIndex ++ );
90114 if (alignedPageReader != null ) {
91115 pageReaderList .add (alignedPageReader );
92116 }
93117 }
94118 }
95119
96- private AlignedPageReader deserializeFromSinglePageChunk () throws IOException {
120+ private AlignedPageReader deserializeFromSinglePageChunk (int pageIndex ) throws IOException {
97121 PageHeader timePageHeader =
98122 PageHeader .deserializeFrom (timeChunkDataBuffer , (Statistics <? extends Serializable >) null );
99123 List <PageHeader > valuePageHeaderList = new ArrayList <>();
@@ -115,7 +139,7 @@ private AlignedPageReader deserializeFromSinglePageChunk() throws IOException {
115139 skipCurrentPage (timePageHeader , valuePageHeaderList );
116140 return null ;
117141 }
118- return constructAlignedPageReader (timePageHeader , valuePageHeaderList );
142+ return constructAlignedPageReader (timePageHeader , valuePageHeaderList , pageIndex );
119143 }
120144
121145 private void skipCurrentPage (PageHeader timePageHeader , List <PageHeader > valuePageHeader ) {
@@ -133,11 +157,17 @@ private void skipCurrentPage(PageHeader timePageHeader, List<PageHeader> valuePa
133157 }
134158
135159 private AlignedPageReader constructAlignedPageReader (
136- PageHeader timePageHeader , List <PageHeader > rawValuePageHeaderList ) throws IOException {
160+ PageHeader timePageHeader , List <PageHeader > rawValuePageHeaderList , int pageIndex )
161+ throws IOException {
137162 IDecryptor decryptor = IDecryptor .getDecryptor (encryptParam );
138163 ByteBuffer timePageData =
139164 ChunkReader .deserializePageData (
140- timePageHeader , timeChunkDataBuffer , timeChunkHeader , decryptor );
165+ timePageHeader ,
166+ timeChunkDataBuffer ,
167+ timeChunkHeader ,
168+ decryptor ,
169+ encryptParam ,
170+ pageIndex );
141171
142172 List <PageHeader > valuePageHeaderList = new ArrayList <>();
143173 LazyLoadPageData [] valuePageDataArray = new LazyLoadPageData [rawValuePageHeaderList .size ()];
@@ -166,7 +196,9 @@ private AlignedPageReader constructAlignedPageReader(
166196 valueChunkDataBufferList .get (i ).array (),
167197 currentPagePosition ,
168198 IUnCompressor .getUnCompressor (valueChunkHeader .getCompressionType ()),
169- encryptParam );
199+ encryptParam ,
200+ pageIndex ,
201+ valueChunkHeader .getChunkOrdinal ());
170202 valueDataTypeList .add (valueChunkHeader .getDataType ());
171203 valueDecoderList .add (
172204 Decoder .getDecoderByType (
0 commit comments