0
# Server-Side Components
1

2
Phoenix's server-side components run within HBase region servers as coprocessors, providing distributed query processing, metadata management, index maintenance, and caching operations. These components enable Phoenix's SQL capabilities while leveraging HBase's scalability.
3

4
## Core Imports
5

6
```java
7
import org.apache.phoenix.coprocessor.*;
8
import org.apache.phoenix.hbase.index.*;
9
import org.apache.hadoop.hbase.coprocessor.*;
10
import org.apache.hadoop.hbase.client.*;
11
import java.io.IOException;
12
```
13

14
## Coprocessor Framework
15

16
### PhoenixCoprocessor
17

18
Base interface for all Phoenix coprocessors providing common functionality and lifecycle management.
19

20
```java{ .api }
21
public interface PhoenixCoprocessor {
22
    // Lifecycle methods
23
    void start(CoprocessorEnvironment env) throws IOException
24
    void stop(CoprocessorEnvironment env) throws IOException
25

26
    // Configuration
27
    Configuration getConfiguration()
28
    HRegionInfo getRegionInfo()
29

30
    // Error handling
31
    void handleException(Exception e) throws IOException
32
}
33
```
34

35
### MetaDataEndpointImpl
36

37
Server-side metadata operations coprocessor handling schema changes and metadata queries.
38

39
```java{ .api }
40
public class MetaDataEndpointImpl extends BaseEndpointCoprocessor
41
                                  implements PhoenixMetaDataCoprocessorProtocol {
42
    // Table operations
43
    public MetaDataMutationResult createTable(List<Mutation> tableMetadata) throws IOException
44
    public MetaDataMutationResult dropTable(byte[] schemaName, byte[] tableName,
45
                                           byte[] parentTableName, long timestamp) throws IOException
46
    public MetaDataMutationResult addColumn(List<Mutation> tableMetadata) throws IOException
47
    public MetaDataMutationResult dropColumn(List<Mutation> tableMetadata) throws IOException
48

49
    // Index operations
50
    public MetaDataMutationResult createIndex(List<Mutation> tableMetadata,
51
                                            byte[] physicalIndexName) throws IOException
52
    public MetaDataMutationResult dropIndex(List<Mutation> tableMetadata,
53
                                          byte[] indexName) throws IOException
54

55
    // Schema operations
56
    public MetaDataMutationResult createSchema(List<Mutation> schemaMetadata) throws IOException
57
    public MetaDataMutationResult dropSchema(List<Mutation> schemaMetadata) throws IOException
58

59
    // Function operations
60
    public MetaDataMutationResult createFunction(List<Mutation> functionMetadata) throws IOException
61
    public MetaDataMutationResult dropFunction(List<Mutation> functionMetadata) throws IOException
62

63
    // Metadata queries
64
    public MetaDataResponse getTable(byte[] tenantId, byte[] schemaName,
65
                                   byte[] tableName, long timestamp) throws IOException
66
    public MetaDataResponse getFunctions(byte[] tenantId, List<byte[]> functionNames,
67
                                       long timestamp) throws IOException
68
    public MetaDataResponse getSchema(byte[] tenantId, byte[] schemaName,
69
                                    long timestamp) throws IOException
70
}
71
```
72

73
**Usage:**
74
```java
75
// Server-side metadata operations are typically invoked through client connections
76
// but can be accessed programmatically in coprocessor contexts
77

78
// Example: Custom coprocessor extending MetaDataEndpointImpl
79
public class CustomMetaDataEndpoint extends MetaDataEndpointImpl {
80
    @Override
81
    public MetaDataMutationResult createTable(List<Mutation> tableMetadata) throws IOException {
82
        // Custom logic before table creation
83
        System.out.println("Creating table with " + tableMetadata.size() + " mutations");
84

85
        // Perform additional validation
86
        for (Mutation mutation : tableMetadata) {
87
            if (mutation instanceof Put) {
88
                Put put = (Put) mutation;
89
                // Validate table metadata
90
                validateTableMetadata(put);
91
            }
92
        }
93

94
        // Call parent implementation
95
        MetaDataMutationResult result = super.createTable(tableMetadata);
96

97
        // Custom logic after table creation
98
        if (result.getMutationCode() == MutationCode.TABLE_ALREADY_EXISTS) {
99
            System.out.println("Table already exists");
100
        } else {
101
            System.out.println("Table created successfully");
102
        }
103

104
        return result;
105
    }
106

107
    private void validateTableMetadata(Put put) throws IOException {
108
        // Custom table metadata validation logic
109
        byte[] tableName = put.getRow();
110
        // Validate table name patterns, quotas, etc.
111
    }
112
}
113
```
114

115
### ServerCachingEndpointImpl
116

117
Server-side caching operations coprocessor for distributed cache management.
118

119
```java{ .api }
120
public class ServerCachingEndpointImpl extends BaseEndpointCoprocessor
121
                                       implements ServerCachingProtocol {
122
    // Cache operations
123
    public boolean addServerCache(byte[] tenantId, byte[] cacheId, byte[] cachePtr,
124
                                byte[] txState, ServerCacheFactory cacheFactory) throws IOException
125
    public boolean removeServerCache(byte[] tenantId, byte[] cacheId) throws IOException
126

127
    // Cache queries
128
    public ServerCache getServerCache(byte[] tenantId, byte[] cacheId) throws IOException
129
    public long getSize(byte[] tenantId, byte[] cacheId) throws IOException
130

131
    // Cache statistics
132
    public ServerCacheStats getCacheStats() throws IOException
133
    public void clearAllCache() throws IOException
134
}
135
```
136

137
**Usage:**
138
```java
139
// Server cache usage is typically handled internally by Phoenix
140
// but can be managed in custom coprocessors
141

142
// Example: Custom caching coprocessor
143
public class CustomCachingEndpoint extends ServerCachingEndpointImpl {
144
    private static final Logger LOG = LoggerFactory.getLogger(CustomCachingEndpoint.class);
145

146
    @Override
147
    public boolean addServerCache(byte[] tenantId, byte[] cacheId, byte[] cachePtr,
148
                                 byte[] txState, ServerCacheFactory cacheFactory) throws IOException {
149
        LOG.info("Adding server cache: tenant={}, cacheId={}, size={}",
150
                Bytes.toString(tenantId), Bytes.toString(cacheId), cachePtr.length);
151

152
        // Custom cache validation or preprocessing
153
        if (cachePtr.length > getMaxCacheSize()) {
154
            throw new IOException("Cache size exceeds maximum allowed size");
155
        }
156

157
        boolean result = super.addServerCache(tenantId, cacheId, cachePtr, txState, cacheFactory);
158

159
        if (result) {
160
            LOG.info("Server cache added successfully");
161
        } else {
162
            LOG.warn("Failed to add server cache");
163
        }
164

165
        return result;
166
    }
167

168
    private int getMaxCacheSize() {
169
        return getConfiguration().getInt("phoenix.cache.maxSize", 104857600); // 100MB default
170
    }
171
}
172
```
173

174
## Index Management
175

176
### IndexBuilder
177

178
Interface for building index updates from primary table mutations.
179

180
```java{ .api }
181
public interface IndexBuilder {
182
    // Index update building
183
    Collection<Pair<Mutation, byte[]>> getIndexUpdate(Put put, IndexMetaData indexMetaData)
184
        throws IOException
185
    Collection<Pair<Mutation, byte[]>> getIndexUpdate(Delete delete, IndexMetaData indexMetaData)
186
        throws IOException
187
    Collection<Pair<Mutation, byte[]>> getIndexUpdateForFilteredRows(
188
        Collection<KeyValue> filtered, IndexMetaData indexMetaData) throws IOException
189

190
    // Builder lifecycle
191
    void setup(RegionCoprocessorEnvironment env) throws IOException
192
    void batchStarted(MiniBatchOperationInProgress<Mutation> miniBatchOp,
193
                     IndexMetaData context) throws IOException
194
    void batchCompleted(MiniBatchOperationInProgress<Mutation> miniBatchOp)
195
}
196
```
197

198
### PhoenixIndexBuilder
199

200
Phoenix-specific implementation of IndexBuilder with support for covered columns and functional indexes.
201

202
```java{ .api }
203
public class PhoenixIndexBuilder implements IndexBuilder {
204
    // Index update generation
205
    public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Put put, IndexMetaData indexMetaData)
206
        throws IOException
207
    public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Delete delete, IndexMetaData indexMetaData)
208
        throws IOException
209

210
    // Phoenix-specific functionality
211
    public Collection<Pair<Mutation, byte[]>> getIndexUpdatesForFilteredRows(
212
        Collection<KeyValue> filtered, IndexMetaData indexMetaData) throws IOException
213

214
    // Configuration
215
    public void setup(RegionCoprocessorEnvironment env) throws IOException
216
    public boolean isEnabled(Mutation m) throws IOException
217
}
218
```
219

220
### IndexWriter
221

222
Interface for writing index updates with support for different writing strategies.
223

224
```java{ .api }
225
public interface IndexWriter {
226
    // Index writing
227
    void write(Collection<Pair<Mutation, byte[]>> indexUpdates) throws IOException
228
    void write(Pair<Mutation, byte[]> indexUpdate, boolean allowLocalUpdates) throws IOException
229

230
    // Writer lifecycle
231
    void setup(RegionCoprocessorEnvironment env) throws IOException
232
    void stop(String why) throws IOException
233

234
    // Error handling
235
    void handleFailure(Collection<Pair<Mutation, byte[]>> attempted,
236
                      Exception cause) throws IOException
237
}
238
```
239

240
### ParallelWriterIndexCommitter
241

242
Parallel index writer implementation for high-throughput index maintenance.
243

244
```java{ .api }
245
public class ParallelWriterIndexCommitter implements IndexWriter {
246
    // Parallel writing configuration
247
    public ParallelWriterIndexCommitter(int numThreads)
248
    public ParallelWriterIndexCommitter(ThreadPoolExecutor pool)
249

250
    // Index writing implementation
251
    public void write(Collection<Pair<Mutation, byte[]>> indexUpdates) throws IOException
252
    public void write(Pair<Mutation, byte[]> indexUpdate, boolean allowLocalUpdates) throws IOException
253

254
    // Performance monitoring
255
    public IndexWriterStats getStats()
256
    public void resetStats()
257
}
258
```
259

260
**Usage:**
261
```java
262
// Custom index builder for specialized indexing logic
263
public class CustomIndexBuilder extends PhoenixIndexBuilder {
264
    @Override
265
    public Collection<Pair<Mutation, byte[]>> getIndexUpdate(Put put, IndexMetaData indexMetaData)
266
            throws IOException {
267
        Collection<Pair<Mutation, byte[]>> updates = super.getIndexUpdate(put, indexMetaData);
268

269
        // Add custom index logic
270
        for (Pair<Mutation, byte[]> update : updates) {
271
            Mutation indexMutation = update.getFirst();
272
            byte[] indexTableName = update.getSecond();
273

274
            // Custom processing for specific indexes
275
            if (Bytes.toString(indexTableName).endsWith("_CUSTOM_IDX")) {
276
                enhanceIndexMutation(indexMutation, put);
277
            }
278
        }
279

280
        return updates;
281
    }
282

283
    private void enhanceIndexMutation(Mutation indexMutation, Put originalPut) {
284
        // Add computed columns, derived values, etc.
285
        if (indexMutation instanceof Put) {
286
            Put indexPut = (Put) indexMutation;
287
            // Add computed timestamp
288
            indexPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("computed_ts"),
289
                             Bytes.toBytes(System.currentTimeMillis()));
290
        }
291
    }
292
}
293

294
// Custom index writer with retry logic
295
public class RetryableIndexWriter extends ParallelWriterIndexCommitter {
296
    private final int maxRetries;
297
    private final long retryDelayMs;
298

299
    public RetryableIndexWriter(int numThreads, int maxRetries, long retryDelayMs) {
300
        super(numThreads);
301
        this.maxRetries = maxRetries;
302
        this.retryDelayMs = retryDelayMs;
303
    }
304

305
    @Override
306
    public void write(Collection<Pair<Mutation, byte[]>> indexUpdates) throws IOException {
307
        int attempts = 0;
308
        IOException lastException = null;
309

310
        while (attempts < maxRetries) {
311
            try {
312
                super.write(indexUpdates);
313
                return; // Success
314
            } catch (IOException e) {
315
                lastException = e;
316
                attempts++;
317

318
                if (attempts < maxRetries) {
319
                    try {
320
                        Thread.sleep(retryDelayMs);
321
                    } catch (InterruptedException ie) {
322
                        Thread.currentThread().interrupt();
323
                        throw new IOException("Interrupted during retry", ie);
324
                    }
325
                }
326
            }
327
        }
328

329
        throw new IOException("Failed to write index updates after " + maxRetries + " attempts",
330
                            lastException);
331
    }
332
}
333
```
334

335
## Region Observer Integration
336

337
### PhoenixRegionObserver
338

339
Phoenix region observer that integrates with HBase region lifecycle for SQL operations.
340

341
```java{ .api }
342
public class PhoenixRegionObserver implements RegionObserver, RegionCoprocessor {
343
    // Region lifecycle
344
    public void start(CoprocessorEnvironment e) throws IOException
345
    public void stop(CoprocessorEnvironment e) throws IOException
346

347
    // Mutation interception
348
    public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
349
                              MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException
350
    public void postBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
351
                               MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException
352

353
    // Scan interception
354
    public RegionScanner preScannerOpen(ObserverContext<RegionCoprocessorEnvironment> c,
355
                                       Scan scan, RegionScanner s) throws IOException
356
    public boolean preScannerNext(ObserverContext<RegionCoprocessorEnvironment> c,
357
                                 InternalScanner s, List<Result> result,
358
                                 int limit, boolean hasNext) throws IOException
359

360
    // Region split/merge handling
361
    public void preSplit(ObserverContext<RegionCoprocessorEnvironment> c,
362
                        byte[] splitRow) throws IOException
363
    public void postSplit(ObserverContext<RegionCoprocessorEnvironment> c,
364
                         Region l, Region r) throws IOException
365
}
366
```
367

368
**Usage:**
369
```java
370
// Custom Phoenix region observer for monitoring and custom logic
371
public class CustomPhoenixRegionObserver extends PhoenixRegionObserver {
372
    private static final Logger LOG = LoggerFactory.getLogger(CustomPhoenixRegionObserver.class);
373
    private Timer mutationTimer;
374

375
    @Override
376
    public void start(CoprocessorEnvironment e) throws IOException {
377
        super.start(e);
378
        mutationTimer = new Timer("MutationTimer", true);
379
        LOG.info("Custom Phoenix Region Observer started");
380
    }
381

382
    @Override
383
    public void preBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
384
                              MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
385
        long startTime = System.currentTimeMillis();
386

387
        // Log large batches
388
        if (miniBatchOp.size() > 1000) {
389
            LOG.info("Processing large batch of {} mutations", miniBatchOp.size());
390
        }
391

392
        // Custom validation or preprocessing
393
        for (int i = 0; i < miniBatchOp.size(); i++) {
394
            Mutation mutation = miniBatchOp.getOperation(i);
395
            if (mutation instanceof Put) {
396
                validatePutMutation((Put) mutation);
397
            }
398
        }
399

400
        // Store timing information
401
        c.getEnvironment().getRegion().getRegionInfo().setEndKey(
402
            Bytes.toBytes(String.valueOf(startTime))
403
        );
404

405
        super.preBatchMutate(c, miniBatchOp);
406
    }
407

408
    @Override
409
    public void postBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
410
                               MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
411
        super.postBatchMutate(c, miniBatchOp);
412

413
        // Calculate and log timing
414
        long duration = System.currentTimeMillis() - extractStartTime(c);
415
        if (duration > 5000) { // Log slow operations
416
            LOG.warn("Slow batch mutation: {} mutations took {}ms",
417
                    miniBatchOp.size(), duration);
418
        }
419
    }
420

421
    private void validatePutMutation(Put put) throws IOException {
422
        // Custom validation logic
423
        NavigableMap<byte[], List<Cell>> familyMap = put.getFamilyCellMap();
424
        for (List<Cell> cells : familyMap.values()) {
425
            for (Cell cell : cells) {
426
                if (cell.getValueLength() > getMaxCellSize()) {
427
                    throw new IOException("Cell value too large: " + cell.getValueLength());
428
                }
429
            }
430
        }
431
    }
432

433
    private long extractStartTime(ObserverContext<RegionCoprocessorEnvironment> c) {
434
        byte[] endKey = c.getEnvironment().getRegion().getRegionInfo().getEndKey();
435
        return Long.parseLong(Bytes.toString(endKey));
436
    }
437

438
    private int getMaxCellSize() {
439
        return getConfiguration().getInt("phoenix.custom.maxCellSize", 10485760); // 10MB
440
    }
441

442
    @Override
443
    public void stop(CoprocessorEnvironment e) throws IOException {
444
        if (mutationTimer != null) {
445
            mutationTimer.cancel();
446
        }
447
        super.stop(e);
448
        LOG.info("Custom Phoenix Region Observer stopped");
449
    }
450
}
451
```
452

453
## Server-Side Query Processing
454

455
### AggregateClient
456

457
Client interface for server-side aggregation operations.
458

459
```java{ .api }
460
public class AggregateClient {
461
    // Aggregation execution
462
    public <T> T aggregate(Scan scan, RowProcessor<T> rowProcessor) throws IOException
463
    public Result[] getRows(Scan scan, int numberOfRows) throws IOException
464

465
    // Aggregate functions
466
    public long rowCount(Scan scan) throws IOException
467
    public <R, S> R median(Scan scan, ColumnInterpreter<R, S, P, Q, W> ci,
468
                          byte[] colFamily, byte[] qualifier) throws IOException
469
    public <R, S> S sum(Scan scan, ColumnInterpreter<R, S, P, Q, W> ci,
470
                       byte[] colFamily, byte[] qualifier) throws IOException
471
    public <R, S> R max(Scan scan, ColumnInterpreter<R, S, P, Q, W> ci,
472
                       byte[] colFamily, byte[] qualifier) throws IOException
473
    public <R, S> R min(Scan scan, ColumnInterpreter<R, S, P, Q, W> ci,
474
                       byte[] colFamily, byte[] qualifier) throws IOException
475
}
476
```
477

478
### HashJoinRegionScanner
479

480
Server-side scanner for hash join operations.
481

482
```java{ .api }
483
public class HashJoinRegionScanner extends BaseRegionScanner {
484
    public HashJoinRegionScanner(RegionScanner scanner, List<Expression> expressions,
485
                                Expression postJoinFilterExpression,
486
                                List<HashJoinInfo> joinInfos, TupleProjector projector)
487

488
    // Scanner implementation
489
    public boolean next(List<Cell> results) throws IOException
490
    public boolean next(List<Cell> results, ScannerContext scannerContext) throws IOException
491
    public void close() throws IOException
492

493
    // Join-specific methods
494
    public long getMaxResultSize()
495
    public RegionInfo getRegionInfo()
496
}
497
```
498

499
**Usage:**
500
```java
501
// Server-side aggregation example
502
public class ServerAggregationExample {
503
    public void performServerAggregation(Connection connection, String tableName) throws IOException, SQLException {
504
        PhoenixConnection phoenixConn = connection.unwrap(PhoenixConnection.class);
505
        ConnectionQueryServices queryServices = phoenixConn.getQueryServices();
506

507
        // Create scan for aggregation
508
        Scan scan = new Scan();
509
        scan.addFamily(Bytes.toBytes("cf"));
510

511
        // Custom row processor for aggregation
512
        RowProcessor<AggregationResult> processor = new RowProcessor<AggregationResult>() {
513
            private long count = 0;
514
            private long sum = 0;
515

516
            @Override
517
            public void process(Result result) {
518
                count++;
519
                Cell[] cells = result.rawCells();
520
                for (Cell cell : cells) {
521
                    if (Bytes.equals(CellUtil.cloneQualifier(cell), Bytes.toBytes("salary"))) {
522
                        sum += Bytes.toLong(CellUtil.cloneValue(cell));
523
                    }
524
                }
525
            }
526

527
            @Override
528
            public AggregationResult getResult() {
529
                return new AggregationResult(count, sum, sum / (double) count);
530
            }
531
        };
532

533
        // Execute server-side aggregation
534
        AggregateClient client = new AggregateClient();
535
        Table table = queryServices.getTable(Bytes.toBytes(tableName));
536
        AggregationResult result = client.aggregate(scan, processor);
537

538
        System.out.println("Count: " + result.getCount());
539
        System.out.println("Sum: " + result.getSum());
540
        System.out.println("Average: " + result.getAverage());
541
    }
542
}
543
```
544

545
## Custom Coprocessor Development
546

547
### Custom Phoenix Coprocessor Template
548

549
```java
550
// Template for developing custom Phoenix coprocessors
551
public abstract class CustomPhoenixCoprocessor implements RegionObserver, RegionCoprocessor {
552
    protected Configuration configuration;
553
    protected RegionCoprocessorEnvironment environment;
554

555
    @Override
556
    public void start(CoprocessorEnvironment env) throws IOException {
557
        if (env instanceof RegionCoprocessorEnvironment) {
558
            this.environment = (RegionCoprocessorEnvironment) env;
559
            this.configuration = env.getConfiguration();
560
            onStart();
561
        }
562
    }
563

564
    @Override
565
    public void stop(CoprocessorEnvironment env) throws IOException {
566
        onStop();
567
    }
568

569
    // Template methods for subclasses
570
    protected abstract void onStart() throws IOException;
571
    protected abstract void onStop() throws IOException;
572

573
    // Utility methods
574
    protected Table getPhoenixTable(String tableName) throws IOException {
575
        return environment.getConnection().getTable(TableName.valueOf(tableName));
576
    }
577

578
    protected void logInfo(String message, Object... args) {
579
        System.out.println(String.format("[CustomCoprocessor] " + message, args));
580
    }
581

582
    protected void logError(String message, Throwable t) {
583
        System.err.println("[CustomCoprocessor] ERROR: " + message);
584
        if (t != null) {
585
            t.printStackTrace();
586
        }
587
    }
588
}
589

590
// Example implementation
591
public class AuditTrailCoprocessor extends CustomPhoenixCoprocessor {
592
    private Table auditTable;
593

594
    @Override
595
    protected void onStart() throws IOException {
596
        auditTable = getPhoenixTable("AUDIT_TRAIL");
597
        logInfo("Audit trail coprocessor started");
598
    }
599

600
    @Override
601
    protected void onStop() throws IOException {
602
        if (auditTable != null) {
603
            auditTable.close();
604
        }
605
        logInfo("Audit trail coprocessor stopped");
606
    }
607

608
    @Override
609
    public void postBatchMutate(ObserverContext<RegionCoprocessorEnvironment> c,
610
                               MiniBatchOperationInProgress<Mutation> miniBatchOp) throws IOException {
611
        try {
612
            // Record audit information
613
            for (int i = 0; i < miniBatchOp.size(); i++) {
614
                Mutation mutation = miniBatchOp.getOperation(i);
615
                recordAuditEntry(mutation);
616
            }
617
        } catch (Exception e) {
618
            logError("Failed to record audit entry", e);
619
            // Don't fail the original operation
620
        }
621
    }
622

623
    private void recordAuditEntry(Mutation mutation) throws IOException {
624
        Put auditPut = new Put(generateAuditRowKey());
625
        auditPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("operation_type"),
626
                          Bytes.toBytes(mutation.getClass().getSimpleName()));
627
        auditPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("row_key"),
628
                          mutation.getRow());
629
        auditPut.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("timestamp"),
630
                          Bytes.toBytes(System.currentTimeMillis()));
631

632
        auditTable.put(auditPut);
633
    }
634

635
    private byte[] generateAuditRowKey() {
636
        return Bytes.toBytes(UUID.randomUUID().toString());
637
    }
638
}
639
```
640

641
## Deployment and Configuration
642

643
### Coprocessor Registration
644

645
```java
646
// Programmatic coprocessor registration
647
public class CoprocessorManager {
648
    public static void registerCustomCoprocessors(Admin admin, String tableName) throws IOException {
649
        TableDescriptor tableDescriptor = admin.getTableDescriptor(TableName.valueOf(tableName));
650
        TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableDescriptor);
651

652
        // Add custom coprocessors
653
        builder.setCoprocessor(CustomPhoenixRegionObserver.class.getName());
654
        builder.setCoprocessor(AuditTrailCoprocessor.class.getName());
655

656
        // Update table
657
        admin.modifyTable(builder.build());
658
        System.out.println("Custom coprocessors registered for table: " + tableName);
659
    }
660

661
    public static void unregisterCoprocessors(Admin admin, String tableName) throws IOException {
662
        TableDescriptor tableDescriptor = admin.getTableDescriptor(TableName.valueOf(tableName));
663
        TableDescriptorBuilder builder = TableDescriptorBuilder.newBuilder(tableDescriptor);
664

665
        // Remove coprocessors
666
        builder.removeCoprocessor(CustomPhoenixRegionObserver.class.getName());
667
        builder.removeCoprocessor(AuditTrailCoprocessor.class.getName());
668

669
        admin.modifyTable(builder.build());
670
        System.out.println("Coprocessors unregistered for table: " + tableName);
671
    }
672
}
673

674
// Usage
675
ConnectionQueryServices queryServices = connection.getQueryServices();
676
Admin admin = queryServices.getAdmin();
677

678
CoprocessorManager.registerCustomCoprocessors(admin, "users");
679
```
680

681
### Configuration Properties
682

683
```xml
684
<!-- hbase-site.xml configuration for Phoenix coprocessors -->
685
<configuration>
686
    <!-- Enable Phoenix coprocessors -->
687
    <property>
688
        <name>hbase.coprocessor.region.classes</name>
689
        <value>org.apache.phoenix.coprocessor.PhoenixRegionObserver</value>
690
    </property>
691

692
    <!-- Phoenix index coprocessors -->
693
    <property>
694
        <name>hbase.coprocessor.wal.classes</name>
695
        <value>org.apache.phoenix.hbase.index.wal.IndexedWALEditCodec</value>
696
    </property>
697

698
    <!-- Custom coprocessor configuration -->
699
    <property>
700
        <name>phoenix.custom.maxCellSize</name>
701
        <value>10485760</value>
702
    </property>
703

704
    <property>
705
        <name>phoenix.cache.maxSize</name>
706
        <value>104857600</value>
707
    </property>
708
</configuration>
709
```